nullStrings) {
+ super(targetCodec, JsonNode.class);
+ this.nullStrings = nullStrings;
+ }
+
+ /**
+ * Whether the input is null.
+ *
+ * This method should be used to inspect external inputs that are meant to be converted to
+ * textual CQL types only (text, varchar and ascii).
+ *
+ *
It always considers the empty string as NOT equivalent to NULL, unless the user clearly
+ * specifies that the empty string is to be considered as NULL, through the
+ * codec.nullStrings setting.
+ *
+ *
Do NOT use this method for non-textual CQL types; use {@link #isNullOrEmpty(JsonNode)}
+ * instead.
+ */
+ protected boolean isNull(JsonNode node) {
+ return node == null
+ || node.isNull()
+ || node.isMissingNode()
+ || (node.isValueNode() && nullStrings.contains(node.asText()));
+ }
+
+ /**
+ * Whether the input is null or empty.
+ *
+ *
This method should be used to inspect external inputs that are meant to be converted to
+ * non-textual CQL types only.
+ *
+ *
It always considers the empty string as equivalent to NULL, which is in compliance with the
+ * documentation of codec.nullStrings: "Note that, regardless of this setting, DSBulk
+ * will always convert empty strings to `null` if the target CQL type is not textual (i.e. not
+ * text, varchar or ascii)."
+ *
+ *
Do NOT use this method for textual CQL types; use {@link #isNull(JsonNode)} instead.
+ */
+ protected boolean isNullOrEmpty(JsonNode node) {
+ return isNull(node) || (node.isValueNode() && node.asText().isEmpty());
+ }
+}
diff --git a/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeConvertingCodecProvider.java b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeConvertingCodecProvider.java
new file mode 100644
index 0000000..442b06f
--- /dev/null
+++ b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeConvertingCodecProvider.java
@@ -0,0 +1,429 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datastax.oss.dsbulk.codecs.text.json;
+
+import static com.datastax.oss.dsbulk.codecs.api.CommonConversionContext.BINARY_FORMAT;
+import static com.datastax.oss.dsbulk.codecs.api.CommonConversionContext.GEO_FORMAT;
+import static com.datastax.oss.dsbulk.codecs.text.TextConversionContext.ALLOW_EXTRA_FIELDS;
+import static com.datastax.oss.dsbulk.codecs.text.TextConversionContext.ALLOW_MISSING_FIELDS;
+import static com.datastax.oss.dsbulk.codecs.text.TextConversionContext.BOOLEAN_INPUT_WORDS;
+import static com.datastax.oss.dsbulk.codecs.text.TextConversionContext.BOOLEAN_NUMBERS;
+import static com.datastax.oss.dsbulk.codecs.text.TextConversionContext.EPOCH;
+import static com.datastax.oss.dsbulk.codecs.text.TextConversionContext.LOCAL_DATE_FORMAT;
+import static com.datastax.oss.dsbulk.codecs.text.TextConversionContext.LOCAL_TIME_FORMAT;
+import static com.datastax.oss.dsbulk.codecs.text.TextConversionContext.NULL_STRINGS;
+import static com.datastax.oss.dsbulk.codecs.text.TextConversionContext.NUMBER_FORMAT;
+import static com.datastax.oss.dsbulk.codecs.text.TextConversionContext.OBJECT_MAPPER;
+import static com.datastax.oss.dsbulk.codecs.text.TextConversionContext.OVERFLOW_STRATEGY;
+import static com.datastax.oss.dsbulk.codecs.text.TextConversionContext.ROUNDING_MODE;
+import static com.datastax.oss.dsbulk.codecs.text.TextConversionContext.TIMESTAMP_FORMAT;
+import static com.datastax.oss.dsbulk.codecs.text.TextConversionContext.TIME_UNIT;
+import static com.datastax.oss.dsbulk.codecs.text.TextConversionContext.TIME_UUID_GENERATOR;
+import static com.datastax.oss.dsbulk.codecs.text.TextConversionContext.TIME_ZONE;
+import static com.datastax.oss.dsbulk.codecs.text.json.JsonCodecUtils.JSON_NODE_TYPE;
+import static com.datastax.oss.protocol.internal.ProtocolConstants.DataType.ASCII;
+import static com.datastax.oss.protocol.internal.ProtocolConstants.DataType.BIGINT;
+import static com.datastax.oss.protocol.internal.ProtocolConstants.DataType.BLOB;
+import static com.datastax.oss.protocol.internal.ProtocolConstants.DataType.BOOLEAN;
+import static com.datastax.oss.protocol.internal.ProtocolConstants.DataType.COUNTER;
+import static com.datastax.oss.protocol.internal.ProtocolConstants.DataType.CUSTOM;
+import static com.datastax.oss.protocol.internal.ProtocolConstants.DataType.DATE;
+import static com.datastax.oss.protocol.internal.ProtocolConstants.DataType.DECIMAL;
+import static com.datastax.oss.protocol.internal.ProtocolConstants.DataType.DOUBLE;
+import static com.datastax.oss.protocol.internal.ProtocolConstants.DataType.DURATION;
+import static com.datastax.oss.protocol.internal.ProtocolConstants.DataType.FLOAT;
+import static com.datastax.oss.protocol.internal.ProtocolConstants.DataType.INET;
+import static com.datastax.oss.protocol.internal.ProtocolConstants.DataType.INT;
+import static com.datastax.oss.protocol.internal.ProtocolConstants.DataType.LIST;
+import static com.datastax.oss.protocol.internal.ProtocolConstants.DataType.MAP;
+import static com.datastax.oss.protocol.internal.ProtocolConstants.DataType.SET;
+import static com.datastax.oss.protocol.internal.ProtocolConstants.DataType.SMALLINT;
+import static com.datastax.oss.protocol.internal.ProtocolConstants.DataType.TIME;
+import static com.datastax.oss.protocol.internal.ProtocolConstants.DataType.TIMESTAMP;
+import static com.datastax.oss.protocol.internal.ProtocolConstants.DataType.TIMEUUID;
+import static com.datastax.oss.protocol.internal.ProtocolConstants.DataType.TINYINT;
+import static com.datastax.oss.protocol.internal.ProtocolConstants.DataType.TUPLE;
+import static com.datastax.oss.protocol.internal.ProtocolConstants.DataType.UDT;
+import static com.datastax.oss.protocol.internal.ProtocolConstants.DataType.UUID;
+import static com.datastax.oss.protocol.internal.ProtocolConstants.DataType.VARCHAR;
+import static com.datastax.oss.protocol.internal.ProtocolConstants.DataType.VARINT;
+
+import com.datastax.oss.driver.api.core.CqlIdentifier;
+import com.datastax.oss.driver.api.core.data.TupleValue;
+import com.datastax.oss.driver.api.core.data.UdtValue;
+import com.datastax.oss.driver.api.core.type.CqlVectorType;
+import com.datastax.oss.driver.api.core.type.CustomType;
+import com.datastax.oss.driver.api.core.type.DataType;
+import com.datastax.oss.driver.api.core.type.DataTypes;
+import com.datastax.oss.driver.api.core.type.ListType;
+import com.datastax.oss.driver.api.core.type.MapType;
+import com.datastax.oss.driver.api.core.type.SetType;
+import com.datastax.oss.driver.api.core.type.TupleType;
+import com.datastax.oss.driver.api.core.type.UserDefinedType;
+import com.datastax.oss.driver.api.core.type.codec.CodecNotFoundException;
+import com.datastax.oss.driver.api.core.type.codec.TypeCodec;
+import com.datastax.oss.driver.api.core.type.codec.TypeCodecs;
+import com.datastax.oss.driver.api.core.type.codec.registry.CodecRegistry;
+import com.datastax.oss.driver.api.core.type.reflect.GenericType;
+import com.datastax.oss.driver.internal.core.type.codec.CqlVectorCodec;
+import com.datastax.oss.driver.internal.core.type.codec.registry.DefaultCodecRegistry;
+import com.datastax.oss.driver.shaded.guava.common.collect.ImmutableList;
+import com.datastax.oss.driver.shaded.guava.common.collect.ImmutableMap;
+import com.datastax.oss.dsbulk.codecs.api.ConversionContext;
+import com.datastax.oss.dsbulk.codecs.api.ConvertingCodec;
+import com.datastax.oss.dsbulk.codecs.api.ConvertingCodecFactory;
+import com.datastax.oss.dsbulk.codecs.api.ConvertingCodecProvider;
+import com.datastax.oss.dsbulk.codecs.text.json.dse.JsonNodeToDateRangeCodec;
+import com.datastax.oss.dsbulk.codecs.text.json.dse.JsonNodeToLineStringCodec;
+import com.datastax.oss.dsbulk.codecs.text.json.dse.JsonNodeToPointCodec;
+import com.datastax.oss.dsbulk.codecs.text.json.dse.JsonNodeToPolygonCodec;
+import com.fasterxml.jackson.databind.JsonNode;
+import edu.umd.cs.findbugs.annotations.NonNull;
+import edu.umd.cs.findbugs.annotations.Nullable;
+import java.time.Instant;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A specialized {@link DefaultCodecRegistry} that is capable of producing {@link ConvertingCodec}s.
+ */
+public class JsonNodeConvertingCodecProvider implements ConvertingCodecProvider {
+
+ private static final Logger LOGGER =
+ LoggerFactory.getLogger(JsonNodeConvertingCodecProvider.class);
+
+ private static final String LINE_STRING_CLASS_NAME =
+ "org.apache.cassandra.db.marshal.LineStringType";
+ private static final String POINT_CLASS_NAME = "org.apache.cassandra.db.marshal.PointType";
+ private static final String POLYGON_CLASS_NAME = "org.apache.cassandra.db.marshal.PolygonType";
+ private static final String DATE_RANGE_CLASS_NAME =
+ "org.apache.cassandra.db.marshal.DateRangeType";
+
+ @NonNull
+ @Override
+ public Optional> maybeProvide(
+ @NonNull DataType cqlType,
+ @NonNull GenericType externalJavaType,
+ @NonNull ConvertingCodecFactory codecFactory,
+ boolean rootCodec) {
+ if (!externalJavaType.equals(JSON_NODE_TYPE)) {
+ return Optional.empty();
+ }
+ ConvertingCodec codec =
+ createJsonNodeConvertingCodec(cqlType, codecFactory, rootCodec);
+ return Optional.ofNullable(codec);
+ }
+
+ @Nullable
+ private ConvertingCodec createJsonNodeConvertingCodec(
+ @NonNull DataType cqlType, @NonNull ConvertingCodecFactory codecFactory, boolean rootCodec) {
+ CodecRegistry codecRegistry = codecFactory.getCodecRegistry();
+ ConversionContext context = codecFactory.getContext();
+ // Don't apply null strings for non-root codecs
+ List nullStrings = rootCodec ? context.getAttribute(NULL_STRINGS) : ImmutableList.of();
+ int cqlTypeCode = cqlType.getProtocolCode();
+ switch (cqlTypeCode) {
+ case ASCII:
+ case VARCHAR:
+ TypeCodec typeCodec = codecRegistry.codecFor(cqlType);
+ return new JsonNodeToStringCodec(
+ typeCodec, context.getAttribute(OBJECT_MAPPER), nullStrings);
+ case BOOLEAN:
+ return new JsonNodeToBooleanCodec(context.getAttribute(BOOLEAN_INPUT_WORDS), nullStrings);
+ case TINYINT:
+ return new JsonNodeToByteCodec(
+ context.getAttribute(NUMBER_FORMAT),
+ context.getAttribute(OVERFLOW_STRATEGY),
+ context.getAttribute(ROUNDING_MODE),
+ context.getAttribute(TIMESTAMP_FORMAT),
+ context.getAttribute(TIME_ZONE),
+ context.getAttribute(TIME_UNIT),
+ context.getAttribute(EPOCH),
+ context.getAttribute(BOOLEAN_INPUT_WORDS),
+ context.getAttribute(BOOLEAN_NUMBERS),
+ nullStrings);
+ case SMALLINT:
+ return new JsonNodeToShortCodec(
+ context.getAttribute(NUMBER_FORMAT),
+ context.getAttribute(OVERFLOW_STRATEGY),
+ context.getAttribute(ROUNDING_MODE),
+ context.getAttribute(TIMESTAMP_FORMAT),
+ context.getAttribute(TIME_ZONE),
+ context.getAttribute(TIME_UNIT),
+ context.getAttribute(EPOCH),
+ context.getAttribute(BOOLEAN_INPUT_WORDS),
+ context.getAttribute(BOOLEAN_NUMBERS),
+ nullStrings);
+ case INT:
+ return new JsonNodeToIntegerCodec(
+ context.getAttribute(NUMBER_FORMAT),
+ context.getAttribute(OVERFLOW_STRATEGY),
+ context.getAttribute(ROUNDING_MODE),
+ context.getAttribute(TIMESTAMP_FORMAT),
+ context.getAttribute(TIME_ZONE),
+ context.getAttribute(TIME_UNIT),
+ context.getAttribute(EPOCH),
+ context.getAttribute(BOOLEAN_INPUT_WORDS),
+ context.getAttribute(BOOLEAN_NUMBERS),
+ nullStrings);
+ case BIGINT:
+ return new JsonNodeToLongCodec(
+ TypeCodecs.BIGINT,
+ context.getAttribute(NUMBER_FORMAT),
+ context.getAttribute(OVERFLOW_STRATEGY),
+ context.getAttribute(ROUNDING_MODE),
+ context.getAttribute(TIMESTAMP_FORMAT),
+ context.getAttribute(TIME_ZONE),
+ context.getAttribute(TIME_UNIT),
+ context.getAttribute(EPOCH),
+ context.getAttribute(BOOLEAN_INPUT_WORDS),
+ context.getAttribute(BOOLEAN_NUMBERS),
+ nullStrings);
+ case COUNTER:
+ return new JsonNodeToLongCodec(
+ TypeCodecs.COUNTER,
+ context.getAttribute(NUMBER_FORMAT),
+ context.getAttribute(OVERFLOW_STRATEGY),
+ context.getAttribute(ROUNDING_MODE),
+ context.getAttribute(TIMESTAMP_FORMAT),
+ context.getAttribute(TIME_ZONE),
+ context.getAttribute(TIME_UNIT),
+ context.getAttribute(EPOCH),
+ context.getAttribute(BOOLEAN_INPUT_WORDS),
+ context.getAttribute(BOOLEAN_NUMBERS),
+ nullStrings);
+ case FLOAT:
+ return new JsonNodeToFloatCodec(
+ context.getAttribute(NUMBER_FORMAT),
+ context.getAttribute(OVERFLOW_STRATEGY),
+ context.getAttribute(ROUNDING_MODE),
+ context.getAttribute(TIMESTAMP_FORMAT),
+ context.getAttribute(TIME_ZONE),
+ context.getAttribute(TIME_UNIT),
+ context.getAttribute(EPOCH),
+ context.getAttribute(BOOLEAN_INPUT_WORDS),
+ context.getAttribute(BOOLEAN_NUMBERS),
+ nullStrings);
+ case DOUBLE:
+ return new JsonNodeToDoubleCodec(
+ context.getAttribute(NUMBER_FORMAT),
+ context.getAttribute(OVERFLOW_STRATEGY),
+ context.getAttribute(ROUNDING_MODE),
+ context.getAttribute(TIMESTAMP_FORMAT),
+ context.getAttribute(TIME_ZONE),
+ context.getAttribute(TIME_UNIT),
+ context.getAttribute(EPOCH),
+ context.getAttribute(BOOLEAN_INPUT_WORDS),
+ context.getAttribute(BOOLEAN_NUMBERS),
+ nullStrings);
+ case VARINT:
+ return new JsonNodeToBigIntegerCodec(
+ context.getAttribute(NUMBER_FORMAT),
+ context.getAttribute(OVERFLOW_STRATEGY),
+ context.getAttribute(ROUNDING_MODE),
+ context.getAttribute(TIMESTAMP_FORMAT),
+ context.getAttribute(TIME_ZONE),
+ context.getAttribute(TIME_UNIT),
+ context.getAttribute(EPOCH),
+ context.getAttribute(BOOLEAN_INPUT_WORDS),
+ context.getAttribute(BOOLEAN_NUMBERS),
+ nullStrings);
+ case DECIMAL:
+ return new JsonNodeToBigDecimalCodec(
+ context.getAttribute(NUMBER_FORMAT),
+ context.getAttribute(OVERFLOW_STRATEGY),
+ context.getAttribute(ROUNDING_MODE),
+ context.getAttribute(TIMESTAMP_FORMAT),
+ context.getAttribute(TIME_ZONE),
+ context.getAttribute(TIME_UNIT),
+ context.getAttribute(EPOCH),
+ context.getAttribute(BOOLEAN_INPUT_WORDS),
+ context.getAttribute(BOOLEAN_NUMBERS),
+ nullStrings);
+ case DATE:
+ return new JsonNodeToLocalDateCodec(
+ context.getAttribute(LOCAL_DATE_FORMAT), context.getAttribute(TIME_ZONE), nullStrings);
+ case TIME:
+ return new JsonNodeToLocalTimeCodec(
+ context.getAttribute(LOCAL_TIME_FORMAT), context.getAttribute(TIME_ZONE), nullStrings);
+ case TIMESTAMP:
+ return new JsonNodeToInstantCodec(
+ context.getAttribute(TIMESTAMP_FORMAT),
+ context.getAttribute(TIME_ZONE),
+ context.getAttribute(EPOCH),
+ nullStrings);
+ case INET:
+ return new JsonNodeToInetAddressCodec(nullStrings);
+ case UUID:
+ {
+ ConvertingCodec instantCodec =
+ codecFactory.createConvertingCodec(DataTypes.TIMESTAMP, GenericType.STRING, false);
+ return new JsonNodeToUUIDCodec(
+ TypeCodecs.UUID,
+ instantCodec,
+ context.getAttribute(TIME_UUID_GENERATOR),
+ nullStrings);
+ }
+ case TIMEUUID:
+ {
+ ConvertingCodec instantCodec =
+ codecFactory.createConvertingCodec(DataTypes.TIMESTAMP, GenericType.STRING, false);
+ return new JsonNodeToUUIDCodec(
+ TypeCodecs.TIMEUUID,
+ instantCodec,
+ context.getAttribute(TIME_UUID_GENERATOR),
+ nullStrings);
+ }
+ case BLOB:
+ return new JsonNodeToBlobCodec(context.getAttribute(BINARY_FORMAT), nullStrings);
+ case DURATION:
+ return new JsonNodeToDurationCodec(nullStrings);
+ case LIST:
+ {
+ DataType elementType = ((ListType) cqlType).getElementType();
+ TypeCodec> collectionCodec = codecRegistry.codecFor(cqlType);
+ @SuppressWarnings("unchecked")
+ ConvertingCodec eltCodec =
+ (ConvertingCodec)
+ createJsonNodeConvertingCodec(elementType, codecFactory, false);
+ return new JsonNodeToListCodec<>(
+ collectionCodec, eltCodec, context.getAttribute(OBJECT_MAPPER), nullStrings);
+ }
+ case SET:
+ {
+ DataType elementType = ((SetType) cqlType).getElementType();
+ TypeCodec> collectionCodec = codecRegistry.codecFor(cqlType);
+ @SuppressWarnings("unchecked")
+ ConvertingCodec eltCodec =
+ (ConvertingCodec)
+ createJsonNodeConvertingCodec(elementType, codecFactory, false);
+ return new JsonNodeToSetCodec<>(
+ collectionCodec, eltCodec, context.getAttribute(OBJECT_MAPPER), nullStrings);
+ }
+ case MAP:
+ {
+ DataType keyType = ((MapType) cqlType).getKeyType();
+ DataType valueType = ((MapType) cqlType).getValueType();
+ TypeCodec> mapCodec = codecRegistry.codecFor(cqlType);
+ ConvertingCodec keyCodec =
+ codecFactory.createConvertingCodec(keyType, GenericType.STRING, false);
+ ConvertingCodec valueCodec =
+ codecFactory.createConvertingCodec(valueType, JSON_NODE_TYPE, false);
+ return new JsonNodeToMapCodec<>(
+ mapCodec, keyCodec, valueCodec, context.getAttribute(OBJECT_MAPPER), nullStrings);
+ }
+ case TUPLE:
+ {
+ TypeCodec tupleCodec = codecRegistry.codecFor(cqlType);
+ ImmutableList.Builder> eltCodecs =
+ new ImmutableList.Builder<>();
+ for (DataType eltType : ((TupleType) cqlType).getComponentTypes()) {
+ ConvertingCodec eltCodec =
+ codecFactory.createConvertingCodec(eltType, JSON_NODE_TYPE, false);
+ eltCodecs.add(Objects.requireNonNull(eltCodec));
+ }
+ return new JsonNodeToTupleCodec(
+ tupleCodec,
+ eltCodecs.build(),
+ context.getAttribute(OBJECT_MAPPER),
+ nullStrings,
+ context.getAttribute(ALLOW_EXTRA_FIELDS),
+ context.getAttribute(ALLOW_MISSING_FIELDS));
+ }
+ case UDT:
+ {
+ TypeCodec udtCodec = codecRegistry.codecFor(cqlType);
+ ImmutableMap.Builder> fieldCodecs =
+ new ImmutableMap.Builder<>();
+ List fieldNames = ((UserDefinedType) cqlType).getFieldNames();
+ List fieldTypes = ((UserDefinedType) cqlType).getFieldTypes();
+ assert (fieldNames.size() == fieldTypes.size());
+
+ for (int idx = 0; idx < fieldNames.size(); idx++) {
+ CqlIdentifier fieldName = fieldNames.get(idx);
+ DataType fieldType = fieldTypes.get(idx);
+ ConvertingCodec fieldCodec =
+ codecFactory.createConvertingCodec(fieldType, JSON_NODE_TYPE, false);
+ fieldCodecs.put(fieldName, Objects.requireNonNull(fieldCodec));
+ }
+ return new JsonNodeToUDTCodec(
+ udtCodec,
+ fieldCodecs.build(),
+ context.getAttribute(OBJECT_MAPPER),
+ nullStrings,
+ context.getAttribute(ALLOW_EXTRA_FIELDS),
+ context.getAttribute(ALLOW_MISSING_FIELDS));
+ }
+ case CUSTOM:
+ {
+ CustomType customType = (CustomType) cqlType;
+ switch (customType.getClassName()) {
+ case POINT_CLASS_NAME:
+ return new JsonNodeToPointCodec(
+ context.getAttribute(OBJECT_MAPPER),
+ context.getAttribute(GEO_FORMAT),
+ nullStrings);
+ case LINE_STRING_CLASS_NAME:
+ return new JsonNodeToLineStringCodec(
+ context.getAttribute(OBJECT_MAPPER),
+ context.getAttribute(GEO_FORMAT),
+ nullStrings);
+ case POLYGON_CLASS_NAME:
+ return new JsonNodeToPolygonCodec(
+ context.getAttribute(OBJECT_MAPPER),
+ context.getAttribute(GEO_FORMAT),
+ nullStrings);
+ case DATE_RANGE_CLASS_NAME:
+ return new JsonNodeToDateRangeCodec(nullStrings);
+ case CqlVectorType.CQLVECTOR_CLASS_NAME:
+ CqlVectorType cqlVectorType = (CqlVectorType) cqlType;
+ // Step 1: create a JSON codec which will take the input JSON nodes and generate
+ // something matching the expected data type
+ ConvertingCodec jsonCodec =
+ createJsonNodeConvertingCodec(cqlVectorType.getSubtype(), codecFactory, false);
+ // Step 2: create a conventional codec which will take instances of the Java type
+ // generated by the JSON codec above and perform standard serde on them.
+ ConvertingCodec standardCodec =
+ codecFactory.createConvertingCodec(
+ cqlVectorType.getSubtype(), jsonCodec.getInternalJavaType(), false);
+ return new JsonNodeToVectorCodec(
+ new CqlVectorCodec(cqlVectorType, standardCodec),
+ jsonCodec,
+ context.getAttribute(OBJECT_MAPPER),
+ nullStrings);
+ }
+ }
+ // fall through
+ default:
+ try {
+ TypeCodec innerCodec = codecRegistry.codecFor(cqlType);
+ LOGGER.warn(
+ String.format(
+ "CQL type %s is not officially supported by this version of DSBulk; "
+ + "JSON literals will be parsed and formatted using registered codec %s",
+ cqlType, innerCodec.getClass().getSimpleName()));
+ return new JsonNodeToUnknownTypeCodec<>(innerCodec, nullStrings);
+ } catch (CodecNotFoundException ignored) {
+ }
+ return null;
+ }
+ }
+}
diff --git a/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToBigDecimalCodec.java b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToBigDecimalCodec.java
new file mode 100644
index 0000000..8cacd9c
--- /dev/null
+++ b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToBigDecimalCodec.java
@@ -0,0 +1,86 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datastax.oss.dsbulk.codecs.text.json;
+
+// import static com.datastax.oss.dsbulk.commons.codecs.json.JsonCodecUtils.JSON_NODE_FACTORY;
+
+import com.datastax.oss.driver.api.core.type.codec.TypeCodecs;
+import com.datastax.oss.dsbulk.codecs.api.format.temporal.TemporalFormat;
+import com.datastax.oss.dsbulk.codecs.api.util.CodecUtils;
+import com.datastax.oss.dsbulk.codecs.api.util.OverflowStrategy;
+import com.fasterxml.jackson.databind.JsonNode;
+import io.netty.util.concurrent.FastThreadLocal;
+import java.math.BigDecimal;
+import java.math.RoundingMode;
+import java.text.NumberFormat;
+import java.time.ZoneId;
+import java.time.ZonedDateTime;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+public class JsonNodeToBigDecimalCodec extends JsonNodeToNumberCodec {
+
+ public JsonNodeToBigDecimalCodec(
+ FastThreadLocal numberFormat,
+ OverflowStrategy overflowStrategy,
+ RoundingMode roundingMode,
+ TemporalFormat temporalFormat,
+ ZoneId timeZone,
+ TimeUnit timeUnit,
+ ZonedDateTime epoch,
+ Map booleanStrings,
+ List booleanNumbers,
+ List nullStrings) {
+ super(
+ TypeCodecs.DECIMAL,
+ numberFormat,
+ overflowStrategy,
+ roundingMode,
+ temporalFormat,
+ timeZone,
+ timeUnit,
+ epoch,
+ booleanStrings,
+ booleanNumbers,
+ nullStrings);
+ }
+
+ @Override
+ public BigDecimal externalToInternal(JsonNode node) {
+ if (isNullOrEmpty(node)) {
+ return null;
+ }
+ if (node.isBigDecimal()) {
+ return node.decimalValue();
+ }
+ Number number;
+ if (node.isNumber()) {
+ number = node.numberValue();
+ } else {
+ number = parseNumber(node);
+ }
+ if (number == null) {
+ return null;
+ }
+ return CodecUtils.toBigDecimal(number);
+ }
+
+ @Override
+ public JsonNode internalToExternal(BigDecimal value) {
+ return value == null ? null : JsonCodecUtils.JSON_NODE_FACTORY.numberNode(value);
+ }
+}
diff --git a/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToBigIntegerCodec.java b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToBigIntegerCodec.java
new file mode 100644
index 0000000..9c9914b
--- /dev/null
+++ b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToBigIntegerCodec.java
@@ -0,0 +1,83 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datastax.oss.dsbulk.codecs.text.json;
+
+import static java.util.stream.Collectors.toList;
+
+import com.datastax.oss.driver.api.core.type.codec.TypeCodecs;
+import com.datastax.oss.dsbulk.codecs.api.format.temporal.TemporalFormat;
+import com.datastax.oss.dsbulk.codecs.api.util.OverflowStrategy;
+import com.fasterxml.jackson.databind.JsonNode;
+import io.netty.util.concurrent.FastThreadLocal;
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.math.RoundingMode;
+import java.text.NumberFormat;
+import java.time.ZoneId;
+import java.time.ZonedDateTime;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+public class JsonNodeToBigIntegerCodec extends JsonNodeToNumberCodec {
+
+ public JsonNodeToBigIntegerCodec(
+ FastThreadLocal numberFormat,
+ OverflowStrategy overflowStrategy,
+ RoundingMode roundingMode,
+ TemporalFormat temporalFormat,
+ ZoneId timeZone,
+ TimeUnit timeUnit,
+ ZonedDateTime epoch,
+ Map booleanStrings,
+ List booleanNumbers,
+ List nullStrings) {
+ super(
+ TypeCodecs.VARINT,
+ numberFormat,
+ overflowStrategy,
+ roundingMode,
+ temporalFormat,
+ timeZone,
+ timeUnit,
+ epoch,
+ booleanStrings,
+ booleanNumbers.stream().map(BigDecimal::toBigInteger).collect(toList()),
+ nullStrings);
+ }
+
+ @Override
+ public BigInteger externalToInternal(JsonNode node) {
+ if (isNullOrEmpty(node)) {
+ return null;
+ }
+ if (node.isBigInteger()) {
+ return node.bigIntegerValue();
+ }
+ Number number;
+ if (node.isNumber()) {
+ number = node.numberValue();
+ } else {
+ number = parseNumber(node);
+ }
+ return narrowNumber(number, BigInteger.class);
+ }
+
+ @Override
+ public JsonNode internalToExternal(BigInteger value) {
+ return value == null ? null : JsonCodecUtils.JSON_NODE_FACTORY.numberNode(value);
+ }
+}
diff --git a/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToBlobCodec.java b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToBlobCodec.java
new file mode 100644
index 0000000..a82b910
--- /dev/null
+++ b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToBlobCodec.java
@@ -0,0 +1,56 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datastax.oss.dsbulk.codecs.text.json;
+
+import com.datastax.oss.driver.api.core.type.codec.TypeCodecs;
+import com.datastax.oss.dsbulk.codecs.api.format.binary.BinaryFormat;
+import com.datastax.oss.dsbulk.codecs.api.util.CodecUtils;
+import com.fasterxml.jackson.databind.JsonNode;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.List;
+
+public class JsonNodeToBlobCodec extends JsonNodeConvertingCodec {
+
+ private final BinaryFormat binaryFormat;
+
+ public JsonNodeToBlobCodec(BinaryFormat binaryFormat, List nullStrings) {
+ super(TypeCodecs.BLOB, nullStrings);
+ this.binaryFormat = binaryFormat;
+ }
+
+ @Override
+ public ByteBuffer externalToInternal(JsonNode node) {
+ // Do not test isNullOrEmpty(), it returns true for empty binary nodes
+ if (isNull(node)) {
+ return null;
+ }
+ if (node.isBinary()) {
+ try {
+ return ByteBuffer.wrap(node.binaryValue());
+ } catch (IOException ignored) {
+ // try as a string below
+ }
+ }
+ String s = node.asText();
+ return CodecUtils.parseByteBuffer(s);
+ }
+
+ @Override
+ public JsonNode internalToExternal(ByteBuffer value) {
+ return JsonCodecUtils.JSON_NODE_FACTORY.textNode(binaryFormat.format(value));
+ }
+}
diff --git a/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToBooleanCodec.java b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToBooleanCodec.java
new file mode 100644
index 0000000..6f1efe7
--- /dev/null
+++ b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToBooleanCodec.java
@@ -0,0 +1,54 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datastax.oss.dsbulk.codecs.text.json;
+
+import static com.datastax.oss.dsbulk.codecs.text.json.JsonCodecUtils.JSON_NODE_FACTORY;
+
+import com.datastax.oss.driver.api.core.type.codec.TypeCodecs;
+import com.fasterxml.jackson.databind.JsonNode;
+import java.util.List;
+import java.util.Map;
+
+public class JsonNodeToBooleanCodec extends JsonNodeConvertingCodec {
+
+ private final Map inputs;
+
+ public JsonNodeToBooleanCodec(Map inputs, List nullStrings) {
+ super(TypeCodecs.BOOLEAN, nullStrings);
+ this.inputs = inputs;
+ }
+
+ @Override
+ public Boolean externalToInternal(JsonNode node) {
+ if (isNullOrEmpty(node)) {
+ return null;
+ }
+ if (node.isBoolean()) {
+ return node.asBoolean();
+ }
+ String s = node.asText();
+ Boolean b = inputs.get(s.toLowerCase());
+ if (b == null) {
+ throw new IllegalArgumentException("Invalid boolean value: " + s);
+ }
+ return b;
+ }
+
+ @Override
+ public JsonNode internalToExternal(Boolean value) {
+ return value == null ? null : JSON_NODE_FACTORY.booleanNode(value);
+ }
+}
diff --git a/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToByteCodec.java b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToByteCodec.java
new file mode 100644
index 0000000..b596ef0
--- /dev/null
+++ b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToByteCodec.java
@@ -0,0 +1,82 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datastax.oss.dsbulk.codecs.text.json;
+
+import static java.util.stream.Collectors.toList;
+
+import com.datastax.oss.driver.api.core.type.codec.TypeCodecs;
+import com.datastax.oss.dsbulk.codecs.api.format.temporal.TemporalFormat;
+import com.datastax.oss.dsbulk.codecs.api.util.OverflowStrategy;
+import com.fasterxml.jackson.databind.JsonNode;
+import io.netty.util.concurrent.FastThreadLocal;
+import java.math.BigDecimal;
+import java.math.RoundingMode;
+import java.text.NumberFormat;
+import java.time.ZoneId;
+import java.time.ZonedDateTime;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+public class JsonNodeToByteCodec extends JsonNodeToNumberCodec {
+
+ public JsonNodeToByteCodec(
+ FastThreadLocal numberFormat,
+ OverflowStrategy overflowStrategy,
+ RoundingMode roundingMode,
+ TemporalFormat temporalFormat,
+ ZoneId timeZone,
+ TimeUnit timeUnit,
+ ZonedDateTime epoch,
+ Map booleanStrings,
+ List booleanNumbers,
+ List nullStrings) {
+ super(
+ TypeCodecs.TINYINT,
+ numberFormat,
+ overflowStrategy,
+ roundingMode,
+ temporalFormat,
+ timeZone,
+ timeUnit,
+ epoch,
+ booleanStrings,
+ booleanNumbers.stream().map(BigDecimal::byteValueExact).collect(toList()),
+ nullStrings);
+ }
+
+ @Override
+ public Byte externalToInternal(JsonNode node) {
+ if (isNullOrEmpty(node)) {
+ return null;
+ }
+ Number number;
+ if (node.isNumber()) {
+ number = node.numberValue();
+ } else {
+ number = parseNumber(node);
+ }
+ if (number == null) {
+ return null;
+ }
+ return narrowNumber(number, Byte.class);
+ }
+
+ @Override
+ public JsonNode internalToExternal(Byte value) {
+ return value == null ? null : JsonCodecUtils.JSON_NODE_FACTORY.numberNode(value);
+ }
+}
diff --git a/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToCollectionCodec.java b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToCollectionCodec.java
new file mode 100644
index 0000000..ba870c5
--- /dev/null
+++ b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToCollectionCodec.java
@@ -0,0 +1,81 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datastax.oss.dsbulk.codecs.text.json;
+
+import com.datastax.oss.driver.api.core.type.codec.TypeCodec;
+import com.datastax.oss.dsbulk.codecs.api.ConvertingCodec;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ArrayNode;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.List;
+import java.util.function.Supplier;
+
+public abstract class JsonNodeToCollectionCodec>
+ extends JsonNodeConvertingCodec {
+
+ private final ConvertingCodec eltCodec;
+ private final Supplier collectionSupplier;
+ private final ObjectMapper objectMapper;
+ private final C emptyCollection;
+
+ JsonNodeToCollectionCodec(
+ TypeCodec collectionCodec,
+ ConvertingCodec eltCodec,
+ ObjectMapper objectMapper,
+ Supplier collectionSupplier,
+ List nullStrings,
+ C emptyCollection) {
+ super(collectionCodec, nullStrings);
+ this.eltCodec = eltCodec;
+ this.objectMapper = objectMapper;
+ this.collectionSupplier = collectionSupplier;
+ this.emptyCollection = emptyCollection;
+ }
+
+ @Override
+ public C externalToInternal(JsonNode node) {
+ if (isNullOrEmpty(node)) {
+ return null;
+ }
+ if (!node.isArray()) {
+ throw new IllegalArgumentException("Expecting ARRAY node, got " + node.getNodeType());
+ }
+ if (node.size() == 0) {
+ return emptyCollection;
+ }
+ Iterator elements = node.elements();
+ C collection = collectionSupplier.get();
+ while (elements.hasNext()) {
+ JsonNode element = elements.next();
+ collection.add(eltCodec.externalToInternal(element));
+ }
+ return collection;
+ }
+
+ @Override
+ public JsonNode internalToExternal(C value) {
+ if (value == null) {
+ return null;
+ }
+ ArrayNode root = objectMapper.createArrayNode();
+ for (E element : value) {
+ root.add(eltCodec.internalToExternal(element));
+ }
+ return root;
+ }
+}
diff --git a/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToDoubleCodec.java b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToDoubleCodec.java
new file mode 100644
index 0000000..047aca3
--- /dev/null
+++ b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToDoubleCodec.java
@@ -0,0 +1,85 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datastax.oss.dsbulk.codecs.text.json;
+
+import static java.util.stream.Collectors.toList;
+
+import com.datastax.oss.driver.api.core.type.codec.TypeCodecs;
+import com.datastax.oss.dsbulk.codecs.api.format.temporal.TemporalFormat;
+import com.datastax.oss.dsbulk.codecs.api.util.OverflowStrategy;
+import com.fasterxml.jackson.databind.JsonNode;
+import io.netty.util.concurrent.FastThreadLocal;
+import java.math.BigDecimal;
+import java.math.RoundingMode;
+import java.text.NumberFormat;
+import java.time.ZoneId;
+import java.time.ZonedDateTime;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+public class JsonNodeToDoubleCodec extends JsonNodeToNumberCodec {
+
+ public JsonNodeToDoubleCodec(
+ FastThreadLocal numberFormat,
+ OverflowStrategy overflowStrategy,
+ RoundingMode roundingMode,
+ TemporalFormat temporalFormat,
+ ZoneId timeZone,
+ TimeUnit timeUnit,
+ ZonedDateTime epoch,
+ Map booleanStrings,
+ List booleanNumbers,
+ List nullStrings) {
+ super(
+ TypeCodecs.DOUBLE,
+ numberFormat,
+ overflowStrategy,
+ roundingMode,
+ temporalFormat,
+ timeZone,
+ timeUnit,
+ epoch,
+ booleanStrings,
+ booleanNumbers.stream().map(BigDecimal::doubleValue).collect(toList()),
+ nullStrings);
+ }
+
+ @Override
+ public Double externalToInternal(JsonNode node) {
+ if (isNullOrEmpty(node)) {
+ return null;
+ }
+ if (node.isDouble()) {
+ return node.doubleValue();
+ }
+ Number number;
+ if (node.isNumber()) {
+ number = node.numberValue();
+ } else {
+ number = parseNumber(node);
+ }
+ if (number == null) {
+ return null;
+ }
+ return narrowNumber(number, Double.class);
+ }
+
+ @Override
+ public JsonNode internalToExternal(Double value) {
+ return value == null ? null : JsonCodecUtils.JSON_NODE_FACTORY.numberNode(value);
+ }
+}
diff --git a/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToDurationCodec.java b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToDurationCodec.java
new file mode 100644
index 0000000..4ba50d3
--- /dev/null
+++ b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToDurationCodec.java
@@ -0,0 +1,42 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datastax.oss.dsbulk.codecs.text.json;
+
+import com.datastax.oss.driver.api.core.data.CqlDuration;
+import com.datastax.oss.driver.api.core.type.codec.TypeCodecs;
+import com.fasterxml.jackson.databind.JsonNode;
+import java.util.List;
+
+public class JsonNodeToDurationCodec extends JsonNodeConvertingCodec {
+
+ public JsonNodeToDurationCodec(List nullStrings) {
+ super(TypeCodecs.DURATION, nullStrings);
+ }
+
+ @Override
+ public CqlDuration externalToInternal(JsonNode node) {
+ if (isNullOrEmpty(node)) {
+ return null;
+ }
+ String s = node.asText();
+ return CqlDuration.from(s);
+ }
+
+ @Override
+ public JsonNode internalToExternal(CqlDuration value) {
+ return value == null ? null : JsonCodecUtils.JSON_NODE_FACTORY.textNode(value.toString());
+ }
+}
diff --git a/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToFloatCodec.java b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToFloatCodec.java
new file mode 100644
index 0000000..e251032
--- /dev/null
+++ b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToFloatCodec.java
@@ -0,0 +1,85 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datastax.oss.dsbulk.codecs.text.json;
+
+import static java.util.stream.Collectors.toList;
+
+import com.datastax.oss.driver.api.core.type.codec.TypeCodecs;
+import com.datastax.oss.dsbulk.codecs.api.format.temporal.TemporalFormat;
+import com.datastax.oss.dsbulk.codecs.api.util.OverflowStrategy;
+import com.fasterxml.jackson.databind.JsonNode;
+import io.netty.util.concurrent.FastThreadLocal;
+import java.math.BigDecimal;
+import java.math.RoundingMode;
+import java.text.NumberFormat;
+import java.time.ZoneId;
+import java.time.ZonedDateTime;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+public class JsonNodeToFloatCodec extends JsonNodeToNumberCodec {
+
+ public JsonNodeToFloatCodec(
+ FastThreadLocal numberFormat,
+ OverflowStrategy overflowStrategy,
+ RoundingMode roundingMode,
+ TemporalFormat temporalFormat,
+ ZoneId timeZone,
+ TimeUnit timeUnit,
+ ZonedDateTime epoch,
+ Map booleanStrings,
+ List booleanNumbers,
+ List nullStrings) {
+ super(
+ TypeCodecs.FLOAT,
+ numberFormat,
+ overflowStrategy,
+ roundingMode,
+ temporalFormat,
+ timeZone,
+ timeUnit,
+ epoch,
+ booleanStrings,
+ booleanNumbers.stream().map(BigDecimal::floatValue).collect(toList()),
+ nullStrings);
+ }
+
+ @Override
+ public Float externalToInternal(JsonNode node) {
+ if (isNullOrEmpty(node)) {
+ return null;
+ }
+ if (node.isFloat()) {
+ return node.floatValue();
+ }
+ Number number;
+ if (node.isNumber()) {
+ number = node.numberValue();
+ } else {
+ number = parseNumber(node);
+ }
+ if (number == null) {
+ return null;
+ }
+ return narrowNumber(number, Float.class);
+ }
+
+ @Override
+ public JsonNode internalToExternal(Float value) {
+ return value == null ? null : JsonCodecUtils.JSON_NODE_FACTORY.numberNode(value);
+ }
+}
diff --git a/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToInetAddressCodec.java b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToInetAddressCodec.java
new file mode 100644
index 0000000..1fcedfa
--- /dev/null
+++ b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToInetAddressCodec.java
@@ -0,0 +1,49 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datastax.oss.dsbulk.codecs.text.json;
+
+import com.datastax.oss.driver.api.core.type.codec.TypeCodecs;
+import com.fasterxml.jackson.databind.JsonNode;
+import java.net.InetAddress;
+import java.util.List;
+
+public class JsonNodeToInetAddressCodec extends JsonNodeConvertingCodec {
+
+ public JsonNodeToInetAddressCodec(List nullStrings) {
+ super(TypeCodecs.INET, nullStrings);
+ }
+
+ @Override
+ public InetAddress externalToInternal(JsonNode node) {
+ if (isNullOrEmpty(node)) {
+ return null;
+ }
+ String s = node.asText();
+ if (s.isEmpty()) {
+ throw new IllegalArgumentException("Cannot create inet address from empty string");
+ }
+ try {
+ return InetAddress.getByName(s);
+ } catch (Exception e) {
+ throw new IllegalArgumentException("Cannot parse inet address: " + s);
+ }
+ }
+
+ @Override
+ public JsonNode internalToExternal(InetAddress value) {
+ return value == null ? null : JsonCodecUtils.JSON_NODE_FACTORY.textNode(value.getHostAddress());
+ }
+}
diff --git a/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToInstantCodec.java b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToInstantCodec.java
new file mode 100644
index 0000000..43f2363
--- /dev/null
+++ b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToInstantCodec.java
@@ -0,0 +1,51 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datastax.oss.dsbulk.codecs.text.json;
+
+import com.datastax.oss.driver.api.core.type.codec.TypeCodecs;
+import com.datastax.oss.dsbulk.codecs.api.format.temporal.TemporalFormat;
+import com.datastax.oss.dsbulk.codecs.api.util.CodecUtils;
+import com.fasterxml.jackson.databind.JsonNode;
+import java.time.Instant;
+import java.time.ZoneId;
+import java.time.ZonedDateTime;
+import java.time.temporal.TemporalAccessor;
+import java.util.List;
+
+public class JsonNodeToInstantCodec extends JsonNodeToTemporalCodec {
+
+ private final ZoneId timeZone;
+ private final ZonedDateTime epoch;
+
+ public JsonNodeToInstantCodec(
+ TemporalFormat temporalFormat,
+ ZoneId timeZone,
+ ZonedDateTime epoch,
+ List nullStrings) {
+ super(TypeCodecs.TIMESTAMP, temporalFormat, nullStrings);
+ this.timeZone = timeZone;
+ this.epoch = epoch;
+ }
+
+ @Override
+ public Instant externalToInternal(JsonNode node) {
+ TemporalAccessor temporal = parseTemporalAccessor(node);
+ if (temporal == null) {
+ return null;
+ }
+ return CodecUtils.toInstant(temporal, timeZone, epoch.toLocalDate());
+ }
+}
diff --git a/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToIntegerCodec.java b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToIntegerCodec.java
new file mode 100644
index 0000000..3257910
--- /dev/null
+++ b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToIntegerCodec.java
@@ -0,0 +1,82 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datastax.oss.dsbulk.codecs.text.json;
+
+import static java.util.stream.Collectors.toList;
+
+import com.datastax.oss.driver.api.core.type.codec.TypeCodecs;
+import com.datastax.oss.dsbulk.codecs.api.format.temporal.TemporalFormat;
+import com.datastax.oss.dsbulk.codecs.api.util.OverflowStrategy;
+import com.fasterxml.jackson.databind.JsonNode;
+import io.netty.util.concurrent.FastThreadLocal;
+import java.math.BigDecimal;
+import java.math.RoundingMode;
+import java.text.NumberFormat;
+import java.time.ZoneId;
+import java.time.ZonedDateTime;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+public class JsonNodeToIntegerCodec extends JsonNodeToNumberCodec {
+
+ public JsonNodeToIntegerCodec(
+ FastThreadLocal numberFormat,
+ OverflowStrategy overflowStrategy,
+ RoundingMode roundingMode,
+ TemporalFormat temporalFormat,
+ ZoneId timeZone,
+ TimeUnit timeUnit,
+ ZonedDateTime epoch,
+ Map booleanStrings,
+ List booleanNumbers,
+ List nullStrings) {
+ super(
+ TypeCodecs.INT,
+ numberFormat,
+ overflowStrategy,
+ roundingMode,
+ temporalFormat,
+ timeZone,
+ timeUnit,
+ epoch,
+ booleanStrings,
+ booleanNumbers.stream().map(BigDecimal::intValueExact).collect(toList()),
+ nullStrings);
+ }
+
+ @Override
+ public Integer externalToInternal(JsonNode node) {
+ if (isNullOrEmpty(node)) {
+ return null;
+ }
+ if (node.isInt()) {
+ return node.intValue();
+ }
+ Number number;
+ if (node.isNumber()) {
+ number = node.numberValue();
+ } else {
+ number = parseNumber(node);
+ }
+ return narrowNumber(number, Integer.class);
+ }
+
+ @Override
+ public JsonNode internalToExternal(Integer value) {
+ return value == null ? null : JsonCodecUtils.JSON_NODE_FACTORY.numberNode(value);
+ }
+}
diff --git a/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToListCodec.java b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToListCodec.java
new file mode 100644
index 0000000..ef6b8a9
--- /dev/null
+++ b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToListCodec.java
@@ -0,0 +1,35 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datastax.oss.dsbulk.codecs.text.json;
+
+import com.datastax.oss.driver.api.core.type.codec.TypeCodec;
+import com.datastax.oss.driver.shaded.guava.common.collect.ImmutableList;
+import com.datastax.oss.dsbulk.codecs.api.ConvertingCodec;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.util.ArrayList;
+import java.util.List;
+
+public class JsonNodeToListCodec extends JsonNodeToCollectionCodec> {
+
+ public JsonNodeToListCodec(
+ TypeCodec> collectionCodec,
+ ConvertingCodec eltCodec,
+ ObjectMapper objectMapper,
+ List nullStrings) {
+ super(collectionCodec, eltCodec, objectMapper, ArrayList::new, nullStrings, ImmutableList.of());
+ }
+}
diff --git a/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToLocalDateCodec.java b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToLocalDateCodec.java
new file mode 100644
index 0000000..1732793
--- /dev/null
+++ b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToLocalDateCodec.java
@@ -0,0 +1,45 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datastax.oss.dsbulk.codecs.text.json;
+
+import com.datastax.oss.driver.api.core.type.codec.TypeCodecs;
+import com.datastax.oss.dsbulk.codecs.api.format.temporal.TemporalFormat;
+import com.datastax.oss.dsbulk.codecs.api.util.CodecUtils;
+import com.fasterxml.jackson.databind.JsonNode;
+import java.time.LocalDate;
+import java.time.ZoneId;
+import java.time.temporal.TemporalAccessor;
+import java.util.List;
+
+public class JsonNodeToLocalDateCodec extends JsonNodeToTemporalCodec {
+
+ private final ZoneId timeZone;
+
+ public JsonNodeToLocalDateCodec(
+ TemporalFormat parser, ZoneId timeZone, List nullStrings) {
+ super(TypeCodecs.DATE, parser, nullStrings);
+ this.timeZone = timeZone;
+ }
+
+ @Override
+ public LocalDate externalToInternal(JsonNode node) {
+ TemporalAccessor temporal = parseTemporalAccessor(node);
+ if (temporal == null) {
+ return null;
+ }
+ return CodecUtils.toLocalDate(temporal, timeZone);
+ }
+}
diff --git a/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToLocalTimeCodec.java b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToLocalTimeCodec.java
new file mode 100644
index 0000000..faafe4c
--- /dev/null
+++ b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToLocalTimeCodec.java
@@ -0,0 +1,45 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datastax.oss.dsbulk.codecs.text.json;
+
+import com.datastax.oss.driver.api.core.type.codec.TypeCodecs;
+import com.datastax.oss.dsbulk.codecs.api.format.temporal.TemporalFormat;
+import com.datastax.oss.dsbulk.codecs.api.util.CodecUtils;
+import com.fasterxml.jackson.databind.JsonNode;
+import java.time.LocalTime;
+import java.time.ZoneId;
+import java.time.temporal.TemporalAccessor;
+import java.util.List;
+
+public class JsonNodeToLocalTimeCodec extends JsonNodeToTemporalCodec {
+
+ private final ZoneId timeZone;
+
+ public JsonNodeToLocalTimeCodec(
+ TemporalFormat parser, ZoneId timeZone, List nullStrings) {
+ super(TypeCodecs.TIME, parser, nullStrings);
+ this.timeZone = timeZone;
+ }
+
+ @Override
+ public LocalTime externalToInternal(JsonNode s) {
+ TemporalAccessor temporal = parseTemporalAccessor(s);
+ if (temporal == null) {
+ return null;
+ }
+ return CodecUtils.toLocalTime(temporal, timeZone);
+ }
+}
diff --git a/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToLongCodec.java b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToLongCodec.java
new file mode 100644
index 0000000..eea7b0a
--- /dev/null
+++ b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToLongCodec.java
@@ -0,0 +1,83 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datastax.oss.dsbulk.codecs.text.json;
+
+import static java.util.stream.Collectors.toList;
+
+import com.datastax.oss.driver.api.core.type.codec.PrimitiveLongCodec;
+import com.datastax.oss.dsbulk.codecs.api.format.temporal.TemporalFormat;
+import com.datastax.oss.dsbulk.codecs.api.util.OverflowStrategy;
+import com.fasterxml.jackson.databind.JsonNode;
+import io.netty.util.concurrent.FastThreadLocal;
+import java.math.BigDecimal;
+import java.math.RoundingMode;
+import java.text.NumberFormat;
+import java.time.ZoneId;
+import java.time.ZonedDateTime;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+public class JsonNodeToLongCodec extends JsonNodeToNumberCodec {
+
+ public JsonNodeToLongCodec(
+ PrimitiveLongCodec targetCodec,
+ FastThreadLocal numberFormat,
+ OverflowStrategy overflowStrategy,
+ RoundingMode roundingMode,
+ TemporalFormat temporalFormat,
+ ZoneId timeZone,
+ TimeUnit timeUnit,
+ ZonedDateTime epoch,
+ Map booleanStrings,
+ List booleanNumbers,
+ List nullStrings) {
+ super(
+ targetCodec,
+ numberFormat,
+ overflowStrategy,
+ roundingMode,
+ temporalFormat,
+ timeZone,
+ timeUnit,
+ epoch,
+ booleanStrings,
+ booleanNumbers.stream().map(BigDecimal::longValueExact).collect(toList()),
+ nullStrings);
+ }
+
+ @Override
+ public Long externalToInternal(JsonNode node) {
+ if (isNullOrEmpty(node)) {
+ return null;
+ }
+ if (node.isLong()) {
+ return node.longValue();
+ }
+ Number number;
+ if (node.isNumber()) {
+ number = node.numberValue();
+ } else {
+ number = parseNumber(node);
+ }
+ return narrowNumber(number, Long.class);
+ }
+
+ @Override
+ public JsonNode internalToExternal(Long value) {
+ return value == null ? null : JsonCodecUtils.JSON_NODE_FACTORY.numberNode(value);
+ }
+}
diff --git a/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToMapCodec.java b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToMapCodec.java
new file mode 100644
index 0000000..1ff573b
--- /dev/null
+++ b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToMapCodec.java
@@ -0,0 +1,84 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datastax.oss.dsbulk.codecs.text.json;
+
+import com.datastax.oss.driver.api.core.type.codec.TypeCodec;
+import com.datastax.oss.driver.shaded.guava.common.collect.ImmutableMap;
+import com.datastax.oss.dsbulk.codecs.api.ConvertingCodec;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+public class JsonNodeToMapCodec extends JsonNodeConvertingCodec> {
+
+ private final ConvertingCodec keyCodec;
+ private final ConvertingCodec valueCodec;
+ private final ObjectMapper objectMapper;
+ private final Map emptyMap;
+
+ public JsonNodeToMapCodec(
+ TypeCodec> collectionCodec,
+ ConvertingCodec keyCodec,
+ ConvertingCodec valueCodec,
+ ObjectMapper objectMapper,
+ List nullStrings) {
+ super(collectionCodec, nullStrings);
+ this.keyCodec = keyCodec;
+ this.valueCodec = valueCodec;
+ this.objectMapper = objectMapper;
+ emptyMap = ImmutableMap.of();
+ }
+
+ @Override
+ public Map externalToInternal(JsonNode node) {
+ if (isNullOrEmpty(node)) {
+ return null;
+ }
+ if (!node.isObject()) {
+ throw new IllegalArgumentException("Expecting OBJECT node, got " + node.getNodeType());
+ }
+ if (node.size() == 0) {
+ return emptyMap;
+ }
+ Map map = new LinkedHashMap<>(node.size());
+ Iterator> it = node.fields();
+ while (it.hasNext()) {
+ Map.Entry entry = it.next();
+ map.put(
+ keyCodec.externalToInternal(entry.getKey()),
+ valueCodec.externalToInternal(entry.getValue()));
+ }
+ return map;
+ }
+
+ @Override
+ public JsonNode internalToExternal(Map map) {
+ if (map == null) {
+ return null;
+ }
+ ObjectNode root = objectMapper.createObjectNode();
+ for (Map.Entry entry : map.entrySet()) {
+ String key = keyCodec.internalToExternal(entry.getKey());
+ JsonNode value = valueCodec.internalToExternal(entry.getValue());
+ root.set(key, value);
+ }
+ return root;
+ }
+}
diff --git a/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToNumberCodec.java b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToNumberCodec.java
new file mode 100644
index 0000000..dbad3e0
--- /dev/null
+++ b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToNumberCodec.java
@@ -0,0 +1,86 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datastax.oss.dsbulk.codecs.text.json;
+
+import com.datastax.oss.driver.api.core.type.codec.TypeCodec;
+import com.datastax.oss.dsbulk.codecs.api.format.temporal.TemporalFormat;
+import com.datastax.oss.dsbulk.codecs.api.util.CodecUtils;
+import com.datastax.oss.dsbulk.codecs.api.util.OverflowStrategy;
+import com.fasterxml.jackson.databind.JsonNode;
+import io.netty.util.concurrent.FastThreadLocal;
+import java.math.RoundingMode;
+import java.text.NumberFormat;
+import java.time.ZoneId;
+import java.time.ZonedDateTime;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+abstract class JsonNodeToNumberCodec extends JsonNodeConvertingCodec {
+
+ private final FastThreadLocal numberFormat;
+ private final OverflowStrategy overflowStrategy;
+ private final RoundingMode roundingMode;
+ private final TemporalFormat temporalFormat;
+ private final ZoneId timeZone;
+ private final TimeUnit timeUnit;
+ private final ZonedDateTime epoch;
+ private final Map booleanStrings;
+ private final List booleanNumbers;
+
+ JsonNodeToNumberCodec(
+ TypeCodec targetCodec,
+ FastThreadLocal numberFormat,
+ OverflowStrategy overflowStrategy,
+ RoundingMode roundingMode,
+ TemporalFormat temporalFormat,
+ ZoneId timeZone,
+ TimeUnit timeUnit,
+ ZonedDateTime epoch,
+ Map booleanStrings,
+ List booleanNumbers,
+ List nullStrings) {
+ super(targetCodec, nullStrings);
+ this.numberFormat = numberFormat;
+ this.overflowStrategy = overflowStrategy;
+ this.roundingMode = roundingMode;
+ this.temporalFormat = temporalFormat;
+ this.timeZone = timeZone;
+ this.timeUnit = timeUnit;
+ this.epoch = epoch;
+ this.booleanStrings = booleanStrings;
+ this.booleanNumbers = booleanNumbers;
+ }
+
+ Number parseNumber(JsonNode node) {
+ if (isNullOrEmpty(node)) {
+ return null;
+ }
+ return CodecUtils.parseNumber(
+ node.asText(),
+ numberFormat.get(),
+ temporalFormat,
+ timeZone,
+ timeUnit,
+ epoch,
+ booleanStrings,
+ booleanNumbers);
+ }
+
+ N narrowNumber(Number number, Class targetClass) {
+ return CodecUtils.narrowNumber(number, targetClass, overflowStrategy, roundingMode);
+ }
+}
diff --git a/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToSetCodec.java b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToSetCodec.java
new file mode 100644
index 0000000..ebee816
--- /dev/null
+++ b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToSetCodec.java
@@ -0,0 +1,42 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datastax.oss.dsbulk.codecs.text.json;
+
+import com.datastax.oss.driver.api.core.type.codec.TypeCodec;
+import com.datastax.oss.driver.shaded.guava.common.collect.ImmutableSet;
+import com.datastax.oss.dsbulk.codecs.api.ConvertingCodec;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Set;
+
+public class JsonNodeToSetCodec extends JsonNodeToCollectionCodec> {
+
+ public JsonNodeToSetCodec(
+ TypeCodec> collectionCodec,
+ ConvertingCodec eltCodec,
+ ObjectMapper objectMapper,
+ List nullStrings) {
+ super(
+ collectionCodec,
+ eltCodec,
+ objectMapper,
+ LinkedHashSet::new,
+ nullStrings,
+ ImmutableSet.of());
+ }
+}
diff --git a/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToShortCodec.java b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToShortCodec.java
new file mode 100644
index 0000000..44a0cef
--- /dev/null
+++ b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToShortCodec.java
@@ -0,0 +1,82 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datastax.oss.dsbulk.codecs.text.json;
+
+import static java.util.stream.Collectors.toList;
+
+import com.datastax.oss.driver.api.core.type.codec.TypeCodecs;
+import com.datastax.oss.dsbulk.codecs.api.format.temporal.TemporalFormat;
+import com.datastax.oss.dsbulk.codecs.api.util.OverflowStrategy;
+import com.fasterxml.jackson.databind.JsonNode;
+import io.netty.util.concurrent.FastThreadLocal;
+import java.math.BigDecimal;
+import java.math.RoundingMode;
+import java.text.NumberFormat;
+import java.time.ZoneId;
+import java.time.ZonedDateTime;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+public class JsonNodeToShortCodec extends JsonNodeToNumberCodec {
+
+ public JsonNodeToShortCodec(
+ FastThreadLocal numberFormat,
+ OverflowStrategy overflowStrategy,
+ RoundingMode roundingMode,
+ TemporalFormat temporalFormat,
+ ZoneId timeZone,
+ TimeUnit timeUnit,
+ ZonedDateTime epoch,
+ Map booleanStrings,
+ List booleanNumbers,
+ List nullStrings) {
+ super(
+ TypeCodecs.SMALLINT,
+ numberFormat,
+ overflowStrategy,
+ roundingMode,
+ temporalFormat,
+ timeZone,
+ timeUnit,
+ epoch,
+ booleanStrings,
+ booleanNumbers.stream().map(BigDecimal::shortValueExact).collect(toList()),
+ nullStrings);
+ }
+
+ @Override
+ public Short externalToInternal(JsonNode node) {
+ if (isNullOrEmpty(node)) {
+ return null;
+ }
+ if (node.isShort()) {
+ return node.shortValue();
+ }
+ Number number;
+ if (node.isNumber()) {
+ number = node.numberValue();
+ } else {
+ number = parseNumber(node);
+ }
+ return narrowNumber(number, Short.class);
+ }
+
+ @Override
+ public JsonNode internalToExternal(Short value) {
+ return value == null ? null : JsonCodecUtils.JSON_NODE_FACTORY.numberNode(value);
+ }
+}
diff --git a/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToStringCodec.java b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToStringCodec.java
new file mode 100644
index 0000000..77bd7fd
--- /dev/null
+++ b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToStringCodec.java
@@ -0,0 +1,58 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datastax.oss.dsbulk.codecs.text.json;
+
+import com.datastax.oss.driver.api.core.type.codec.TypeCodec;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.util.List;
+
+public class JsonNodeToStringCodec extends JsonNodeConvertingCodec {
+
+ private final ObjectMapper objectMapper;
+
+ public JsonNodeToStringCodec(
+ TypeCodec innerCodec, ObjectMapper objectMapper, List nullStrings) {
+ super(innerCodec, nullStrings);
+ this.objectMapper = objectMapper;
+ }
+
+ @Override
+ public String externalToInternal(JsonNode node) {
+ // DAT-297: do not convert empty strings to null so do not use isNullOrEmpty() here
+ if (isNull(node)) {
+ return null;
+ }
+ if (node.isContainerNode()) {
+ try {
+ return objectMapper.writeValueAsString(node);
+ } catch (JsonProcessingException e) {
+ throw new IllegalArgumentException("Cannot deserialize node " + node, e);
+ }
+ } else {
+ return node.asText();
+ }
+ }
+
+ @Override
+ public JsonNode internalToExternal(String value) {
+ if (value == null) {
+ return null;
+ }
+ return objectMapper.getNodeFactory().textNode(value);
+ }
+}
diff --git a/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToTemporalCodec.java b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToTemporalCodec.java
new file mode 100644
index 0000000..ae869a3
--- /dev/null
+++ b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToTemporalCodec.java
@@ -0,0 +1,90 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datastax.oss.dsbulk.codecs.text.json;
+
+import com.datastax.oss.driver.api.core.type.codec.TypeCodec;
+import com.datastax.oss.dsbulk.codecs.api.format.temporal.NumericTemporalFormat;
+import com.datastax.oss.dsbulk.codecs.api.format.temporal.TemporalFormat;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.node.NumericNode;
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.time.temporal.TemporalAccessor;
+import java.util.List;
+
+public abstract class JsonNodeToTemporalCodec
+ extends JsonNodeConvertingCodec {
+
+ final TemporalFormat temporalFormat;
+
+ JsonNodeToTemporalCodec(
+ TypeCodec targetCodec, TemporalFormat temporalFormat, List nullStrings) {
+ super(targetCodec, nullStrings);
+ this.temporalFormat = temporalFormat;
+ }
+
+ @Override
+ public JsonNode internalToExternal(T value) {
+ if (value == null) {
+ return null;
+ } else if (temporalFormat instanceof NumericTemporalFormat) {
+ Number n = ((NumericTemporalFormat) temporalFormat).temporalToNumber(value);
+ if (n == null) {
+ return null;
+ }
+ if (n instanceof Byte) {
+ return JsonCodecUtils.JSON_NODE_FACTORY.numberNode((Byte) n);
+ }
+ if (n instanceof Short) {
+ return JsonCodecUtils.JSON_NODE_FACTORY.numberNode((Short) n);
+ }
+ if (n instanceof Integer) {
+ return JsonCodecUtils.JSON_NODE_FACTORY.numberNode((Integer) n);
+ }
+ if (n instanceof Long) {
+ return JsonCodecUtils.JSON_NODE_FACTORY.numberNode((Long) n);
+ }
+ if (n instanceof Float) {
+ return JsonCodecUtils.JSON_NODE_FACTORY.numberNode((Float) n);
+ }
+ if (n instanceof Double) {
+ return JsonCodecUtils.JSON_NODE_FACTORY.numberNode((Double) n);
+ }
+ if (n instanceof BigInteger) {
+ return JsonCodecUtils.JSON_NODE_FACTORY.numberNode((BigInteger) n);
+ }
+ if (n instanceof BigDecimal) {
+ return JsonCodecUtils.JSON_NODE_FACTORY.numberNode((BigDecimal) n);
+ }
+ return JsonCodecUtils.JSON_NODE_FACTORY.textNode(n.toString());
+ } else {
+ return JsonCodecUtils.JSON_NODE_FACTORY.textNode(temporalFormat.format(value));
+ }
+ }
+
+ TemporalAccessor parseTemporalAccessor(JsonNode node) {
+ if (isNullOrEmpty(node)) {
+ return null;
+ }
+ if (node instanceof NumericNode && temporalFormat instanceof NumericTemporalFormat) {
+ Number n = node.numberValue();
+ return ((NumericTemporalFormat) temporalFormat).numberToTemporal(n);
+ } else {
+ String s = node.asText();
+ return temporalFormat.parse(s);
+ }
+ }
+}
diff --git a/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToTupleCodec.java b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToTupleCodec.java
new file mode 100644
index 0000000..09c55da
--- /dev/null
+++ b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToTupleCodec.java
@@ -0,0 +1,90 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datastax.oss.dsbulk.codecs.text.json;
+
+import com.datastax.oss.driver.api.core.data.TupleValue;
+import com.datastax.oss.driver.api.core.type.TupleType;
+import com.datastax.oss.driver.api.core.type.codec.TypeCodec;
+import com.datastax.oss.dsbulk.codecs.api.ConvertingCodec;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ArrayNode;
+import java.util.List;
+
+public class JsonNodeToTupleCodec extends JsonNodeConvertingCodec {
+
+ private final TupleType definition;
+ private final List> eltCodecs;
+ private final ObjectMapper objectMapper;
+ private final boolean allowExtraFields;
+ private final boolean allowMissingFields;
+
+ public JsonNodeToTupleCodec(
+ TypeCodec tupleCodec,
+ List> eltCodecs,
+ ObjectMapper objectMapper,
+ List nullStrings,
+ boolean allowExtraFields,
+ boolean allowMissingFields) {
+ super(tupleCodec, nullStrings);
+ this.eltCodecs = eltCodecs;
+ definition = (TupleType) tupleCodec.getCqlType();
+ this.objectMapper = objectMapper;
+ this.allowExtraFields = allowExtraFields;
+ this.allowMissingFields = allowMissingFields;
+ }
+
+ @Override
+ public TupleValue externalToInternal(JsonNode node) {
+ if (isNullOrEmpty(node)) {
+ return null;
+ }
+ if (!node.isArray()) {
+ throw new IllegalArgumentException("Expecting ARRAY node, got " + node.getNodeType());
+ }
+ int tupleSize = definition.getComponentTypes().size();
+ int nodeSize = node.size();
+ if (nodeSize > tupleSize && !allowExtraFields) {
+ throw JsonSchemaMismatchException.arraySizeGreaterThanTupleSize(tupleSize, nodeSize);
+ }
+ if (nodeSize < tupleSize && !allowMissingFields) {
+ throw JsonSchemaMismatchException.arraySizeLesserThanTupleSize(tupleSize, nodeSize);
+ }
+ TupleValue tuple = definition.newValue();
+ for (int i = 0; i < tupleSize && i < nodeSize; i++) {
+ ConvertingCodec eltCodec = eltCodecs.get(i);
+ Object o = eltCodec.externalToInternal(node.get(i));
+ tuple = tuple.set(i, o, eltCodec.getInternalJavaType());
+ }
+ return tuple;
+ }
+
+ @Override
+ public JsonNode internalToExternal(TupleValue tuple) {
+ if (tuple == null) {
+ return null;
+ }
+ ArrayNode root = objectMapper.createArrayNode();
+ int size = definition.getComponentTypes().size();
+ for (int i = 0; i < size; i++) {
+ ConvertingCodec eltCodec = eltCodecs.get(i);
+ Object o = tuple.get(i, eltCodec.getInternalJavaType());
+ JsonNode element = eltCodec.internalToExternal(o);
+ root.add(element);
+ }
+ return root;
+ }
+}
diff --git a/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToUDTCodec.java b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToUDTCodec.java
new file mode 100644
index 0000000..bc2d69e
--- /dev/null
+++ b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToUDTCodec.java
@@ -0,0 +1,143 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datastax.oss.dsbulk.codecs.text.json;
+
+import com.datastax.oss.driver.api.core.CqlIdentifier;
+import com.datastax.oss.driver.api.core.data.UdtValue;
+import com.datastax.oss.driver.api.core.type.UserDefinedType;
+import com.datastax.oss.driver.api.core.type.codec.TypeCodec;
+import com.datastax.oss.driver.shaded.guava.common.collect.Iterators;
+import com.datastax.oss.driver.shaded.guava.common.collect.Sets;
+import com.datastax.oss.dsbulk.codecs.api.ConvertingCodec;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import java.util.Iterator;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class JsonNodeToUDTCodec extends JsonNodeConvertingCodec {
+
+ private final Map> fieldCodecs;
+ private final UserDefinedType definition;
+ private final ObjectMapper objectMapper;
+ private final boolean allowExtraFields;
+ private final boolean allowMissingFields;
+
+ public JsonNodeToUDTCodec(
+ TypeCodec udtCodec,
+ Map> fieldCodecs,
+ ObjectMapper objectMapper,
+ List nullStrings,
+ boolean allowExtraFields,
+ boolean allowMissingFields) {
+ super(udtCodec, nullStrings);
+ this.fieldCodecs = fieldCodecs;
+ definition = (UserDefinedType) udtCodec.getCqlType();
+ this.objectMapper = objectMapper;
+ this.allowExtraFields = allowExtraFields;
+ this.allowMissingFields = allowMissingFields;
+ }
+
+ @Override
+ public UdtValue externalToInternal(JsonNode node) {
+ if (isNullOrEmpty(node)) {
+ return null;
+ }
+ if (!(node.isObject() || node.isArray())) {
+ throw new IllegalArgumentException(
+ "Expecting OBJECT or ARRAY node, got " + node.getNodeType());
+ }
+ if (node.size() == 0 && allowMissingFields) {
+ return definition.newValue();
+ }
+ UdtValue value = definition.newValue();
+ if (node.isObject()) {
+ checkJsonObject(node);
+ for (CqlIdentifier field : definition.getFieldNames()) {
+ String name = field.asInternal();
+ if (node.has(name)) {
+ ConvertingCodec fieldCodec = fieldCodecs.get(field);
+ Object o = fieldCodec.externalToInternal(node.get(name));
+ value = value.set(field, o, fieldCodec.getInternalJavaType());
+ }
+ }
+ } else {
+ checkJsonArray(node);
+ // The field iteration order is deterministic
+ Iterator fields = definition.getFieldNames().iterator();
+ for (int i = 0; i < node.size() && fields.hasNext(); i++) {
+ JsonNode element = node.get(i);
+ CqlIdentifier field = fields.next();
+ ConvertingCodec fieldCodec = fieldCodecs.get(field);
+ Object o = fieldCodec.externalToInternal(element);
+ value = value.set(i, o, fieldCodec.getInternalJavaType());
+ }
+ }
+ return value;
+ }
+
+ private void checkJsonObject(JsonNode node) {
+ Set udtFieldNames =
+ definition.getFieldNames().stream()
+ .map(CqlIdentifier::asInternal)
+ .collect(Collectors.toCollection(LinkedHashSet::new));
+ Set nodeFieldNames = new LinkedHashSet<>();
+ Iterators.addAll(nodeFieldNames, node.fieldNames());
+ if (!udtFieldNames.equals(nodeFieldNames)) {
+ Set extraneous = Sets.difference(nodeFieldNames, udtFieldNames);
+ Set missing = Sets.difference(udtFieldNames, nodeFieldNames);
+ boolean hasExtras = !allowExtraFields && !extraneous.isEmpty();
+ boolean hasMissing = !allowMissingFields && !missing.isEmpty();
+ if (hasMissing && hasExtras) {
+ throw JsonSchemaMismatchException.objectHasMissingAndExtraneousFields(extraneous, missing);
+ } else if (hasExtras) {
+ throw JsonSchemaMismatchException.objectHasExtraneousFields(extraneous);
+ } else if (hasMissing) {
+ throw JsonSchemaMismatchException.objectHasMissingFields(missing);
+ }
+ }
+ }
+
+ private void checkJsonArray(JsonNode node) {
+ int udtSize = definition.getFieldNames().size();
+ int nodeSize = node.size();
+ if (nodeSize > udtSize && !allowExtraFields) {
+ throw JsonSchemaMismatchException.arraySizeGreaterThanUDTSize(udtSize, nodeSize);
+ }
+ if (nodeSize < udtSize && !allowMissingFields) {
+ throw JsonSchemaMismatchException.arraySizeLesserThanUDTSize(udtSize, nodeSize);
+ }
+ }
+
+ @Override
+ public JsonNode internalToExternal(UdtValue value) {
+ if (value == null) {
+ return null;
+ }
+ ObjectNode root = objectMapper.createObjectNode();
+ for (CqlIdentifier field : definition.getFieldNames()) {
+ ConvertingCodec eltCodec = fieldCodecs.get(field);
+ Object o = value.get(field, eltCodec.getInternalJavaType());
+ JsonNode node = eltCodec.internalToExternal(o);
+ root.set(field.asInternal(), node);
+ }
+ return root;
+ }
+}
diff --git a/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToUUIDCodec.java b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToUUIDCodec.java
new file mode 100644
index 0000000..735e9dc
--- /dev/null
+++ b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToUUIDCodec.java
@@ -0,0 +1,54 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datastax.oss.dsbulk.codecs.text.json;
+
+import com.datastax.oss.driver.api.core.type.codec.TypeCodec;
+import com.datastax.oss.dsbulk.codecs.api.ConvertingCodec;
+import com.datastax.oss.dsbulk.codecs.api.util.CodecUtils;
+import com.datastax.oss.dsbulk.codecs.api.util.TimeUUIDGenerator;
+import com.fasterxml.jackson.databind.JsonNode;
+import java.time.Instant;
+import java.util.List;
+import java.util.UUID;
+
+public class JsonNodeToUUIDCodec extends JsonNodeConvertingCodec {
+
+ private final ConvertingCodec instantCodec;
+ private final TimeUUIDGenerator generator;
+
+ public JsonNodeToUUIDCodec(
+ TypeCodec targetCodec,
+ ConvertingCodec instantCodec,
+ TimeUUIDGenerator generator,
+ List nullStrings) {
+ super(targetCodec, nullStrings);
+ this.instantCodec = instantCodec;
+ this.generator = generator;
+ }
+
+ @Override
+ public UUID externalToInternal(JsonNode node) {
+ if (isNullOrEmpty(node)) {
+ return null;
+ }
+ return CodecUtils.parseUUID(node.asText(), instantCodec, generator);
+ }
+
+ @Override
+ public JsonNode internalToExternal(UUID value) {
+ return value == null ? null : JsonCodecUtils.JSON_NODE_FACTORY.textNode(value.toString());
+ }
+}
diff --git a/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToUnknownTypeCodec.java b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToUnknownTypeCodec.java
new file mode 100644
index 0000000..de52429
--- /dev/null
+++ b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToUnknownTypeCodec.java
@@ -0,0 +1,51 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datastax.oss.dsbulk.codecs.text.json;
+
+import static com.datastax.oss.dsbulk.codecs.text.json.JsonCodecUtils.JSON_NODE_FACTORY;
+
+import com.datastax.oss.driver.api.core.type.codec.TypeCodec;
+import com.fasterxml.jackson.databind.JsonNode;
+import java.util.List;
+
+public class JsonNodeToUnknownTypeCodec extends JsonNodeConvertingCodec {
+
+ public JsonNodeToUnknownTypeCodec(TypeCodec targetCodec, List nullStrings) {
+ super(targetCodec, nullStrings);
+ }
+
+ @Override
+ public T externalToInternal(JsonNode node) {
+ if (isNullOrEmpty(node)) {
+ return null;
+ }
+ return getInternalCodec().parse(node.asText());
+ }
+
+ @Override
+ public JsonNode internalToExternal(T o) {
+ if (o == null) {
+ return null;
+ }
+ String s = getInternalCodec().format(o);
+ // most codecs usually format null/empty values using the CQL keyword "NULL",
+ // but some others may choose to return a null string.
+ if (s.equalsIgnoreCase("NULL")) {
+ return null;
+ }
+ return JSON_NODE_FACTORY.textNode(s);
+ }
+}
diff --git a/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToVectorCodec.java b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToVectorCodec.java
new file mode 100644
index 0000000..b0a7f8c
--- /dev/null
+++ b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonNodeToVectorCodec.java
@@ -0,0 +1,60 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datastax.oss.dsbulk.codecs.text.json;
+
+import com.datastax.oss.driver.api.core.data.CqlVector;
+import com.datastax.oss.driver.internal.core.type.codec.CqlVectorCodec;
+import com.datastax.oss.dsbulk.codecs.api.ConvertingCodec;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ArrayNode;
+import java.util.Iterator;
+import java.util.List;
+
+public class JsonNodeToVectorCodec extends JsonNodeConvertingCodec> {
+
+ private final ConvertingCodec subtypeCodec;
+ private final ObjectMapper objectMapper;
+
+ public JsonNodeToVectorCodec(
+ CqlVectorCodec targetCodec,
+ ConvertingCodec subtypeCodec,
+ ObjectMapper objectMapper,
+ List nullStrings) {
+ super(targetCodec, nullStrings);
+ this.subtypeCodec = subtypeCodec;
+ this.objectMapper = objectMapper;
+ }
+
+ @Override
+ public CqlVector externalToInternal(JsonNode jsonNode) {
+ if (jsonNode == null || !jsonNode.isArray()) return null;
+ CqlVector.Builder builder = CqlVector.builder();
+ for (Iterator it = jsonNode.elements(); it.hasNext(); )
+ builder.add(subtypeCodec.externalToInternal(it.next()));
+ return builder.build();
+ }
+
+ @Override
+ public JsonNode internalToExternal(CqlVector value) {
+ if (value == null) return null;
+ ArrayNode root = objectMapper.createArrayNode();
+ for (SubtypeT element : value.getValues()) {
+ root.add(subtypeCodec.internalToExternal(element));
+ }
+ return root;
+ }
+}
diff --git a/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonSchemaMismatchException.java b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonSchemaMismatchException.java
new file mode 100644
index 0000000..4b6cfd8
--- /dev/null
+++ b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/JsonSchemaMismatchException.java
@@ -0,0 +1,125 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datastax.oss.dsbulk.codecs.text.json;
+
+import edu.umd.cs.findbugs.annotations.NonNull;
+import java.util.Set;
+
+class JsonSchemaMismatchException extends IllegalArgumentException {
+
+ private JsonSchemaMismatchException(String message) {
+ super(message);
+ }
+
+ @NonNull
+ static JsonSchemaMismatchException objectHasMissingAndExtraneousFields(
+ Set extraneous, Set missing) {
+ StringBuilder msg =
+ new StringBuilder("JSON object does not match UDT definition: found ")
+ .append(extraneous.size())
+ .append(" extraneous field");
+ if (extraneous.size() > 1) {
+ msg.append('s');
+ }
+ msg.append(": '")
+ .append(String.join("', '", extraneous))
+ .append("' and ")
+ .append(missing.size())
+ .append(" missing field");
+ if (missing.size() > 1) {
+ msg.append('s');
+ }
+ msg.append(": '")
+ .append(String.join("', '", missing))
+ .append("' (set schema.allowExtraFields to true to allow ")
+ .append("JSON objects to contain fields not present in the UDT definition and ")
+ .append("set schema.allowMissingFields to true to allow ")
+ .append("JSON objects to lack of fields present in the UDT definition).");
+ return new JsonSchemaMismatchException(msg.toString());
+ }
+
+ @NonNull
+ static JsonSchemaMismatchException objectHasExtraneousFields(Set extraneous) {
+ StringBuilder msg =
+ new StringBuilder("JSON object does not match UDT definition: found ")
+ .append(extraneous.size())
+ .append(" extraneous field");
+ if (extraneous.size() > 1) {
+ msg.append('s');
+ }
+ msg.append(": '")
+ .append(String.join("', '", extraneous))
+ .append("' (set schema.allowExtraFields to true to allow ")
+ .append("JSON objects to contain fields not present in the UDT definition).");
+ return new JsonSchemaMismatchException(msg.toString());
+ }
+
+ @NonNull
+ static JsonSchemaMismatchException objectHasMissingFields(Set missing) {
+ StringBuilder msg =
+ new StringBuilder("JSON object does not match UDT definition: found ")
+ .append(missing.size())
+ .append(" missing field");
+ if (missing.size() > 1) {
+ msg.append('s');
+ }
+ msg.append(": '")
+ .append(String.join("', '", missing))
+ .append("' (set schema.allowMissingFields to true to allow ")
+ .append("JSON objects to lack of fields present in the UDT definition).");
+ return new JsonSchemaMismatchException(msg.toString());
+ }
+
+ @NonNull
+ static JsonSchemaMismatchException arraySizeGreaterThanUDTSize(int udtSize, int nodeSize) {
+ return new JsonSchemaMismatchException(
+ String.format(
+ "JSON array does not match UDT definition: expecting %d elements, got %d "
+ + "(set schema.allowExtraFields to true to allow "
+ + "JSON arrays to contain more elements than the UDT definition).",
+ udtSize, nodeSize));
+ }
+
+ @NonNull
+ static JsonSchemaMismatchException arraySizeLesserThanUDTSize(int udtSize, int nodeSize) {
+ return new JsonSchemaMismatchException(
+ String.format(
+ "JSON array does not match UDT definition: expecting %d elements, got %d "
+ + "(set schema.allowMissingFields to true to allow "
+ + "JSON arrays to contain fewer elements than the UDT definition).",
+ udtSize, nodeSize));
+ }
+
+ @NonNull
+ static JsonSchemaMismatchException arraySizeLesserThanTupleSize(int tupleSize, int nodeSize) {
+ return new JsonSchemaMismatchException(
+ String.format(
+ "JSON array does not match tuple definition: expecting %d elements, got %d "
+ + "(set schema.allowMissingFields to true to allow "
+ + "JSON arrays to contain fewer elements than the tuple definition).",
+ tupleSize, nodeSize));
+ }
+
+ @NonNull
+ static JsonSchemaMismatchException arraySizeGreaterThanTupleSize(int tupleSize, int nodeSize) {
+ return new JsonSchemaMismatchException(
+ String.format(
+ "JSON array does not match tuple definition: expecting %d elements, got %d "
+ + "(set schema.allowExtraFields to true to allow "
+ + "JSON arrays to contain more elements than the tuple definition).",
+ tupleSize, nodeSize));
+ }
+}
diff --git a/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/dse/JsonNodeToDateRangeCodec.java b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/dse/JsonNodeToDateRangeCodec.java
new file mode 100644
index 0000000..352eda9
--- /dev/null
+++ b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/dse/JsonNodeToDateRangeCodec.java
@@ -0,0 +1,45 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datastax.oss.dsbulk.codecs.text.json.dse;
+
+import static com.datastax.oss.dsbulk.codecs.text.json.JsonCodecUtils.JSON_NODE_FACTORY;
+
+import com.datastax.dse.driver.api.core.data.time.DateRange;
+import com.datastax.dse.driver.internal.core.type.codec.time.DateRangeCodec;
+import com.datastax.oss.dsbulk.codecs.api.util.CodecUtils;
+import com.datastax.oss.dsbulk.codecs.text.json.JsonNodeConvertingCodec;
+import com.fasterxml.jackson.databind.JsonNode;
+import java.util.List;
+
+public class JsonNodeToDateRangeCodec extends JsonNodeConvertingCodec {
+
+ public JsonNodeToDateRangeCodec(List nullStrings) {
+ super(new DateRangeCodec(), nullStrings);
+ }
+
+ @Override
+ public DateRange externalToInternal(JsonNode node) {
+ if (isNullOrEmpty(node)) {
+ return null;
+ }
+ return CodecUtils.parseDateRange(node.asText());
+ }
+
+ @Override
+ public JsonNode internalToExternal(DateRange value) {
+ return value == null ? null : JSON_NODE_FACTORY.textNode(value.toString());
+ }
+}
diff --git a/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/dse/JsonNodeToGeometryCodec.java b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/dse/JsonNodeToGeometryCodec.java
new file mode 100644
index 0000000..0506338
--- /dev/null
+++ b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/dse/JsonNodeToGeometryCodec.java
@@ -0,0 +1,95 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datastax.oss.dsbulk.codecs.text.json.dse;
+
+import com.datastax.dse.driver.api.core.data.geometry.Geometry;
+import com.datastax.oss.driver.api.core.type.codec.TypeCodec;
+import com.datastax.oss.dsbulk.codecs.api.format.geo.GeoFormat;
+import com.datastax.oss.dsbulk.codecs.api.format.geo.JsonGeoFormat;
+import com.datastax.oss.dsbulk.codecs.text.json.JsonNodeConvertingCodec;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import edu.umd.cs.findbugs.annotations.NonNull;
+import java.io.IOException;
+import java.util.List;
+
+public abstract class JsonNodeToGeometryCodec
+ extends JsonNodeConvertingCodec {
+
+ protected final ObjectMapper objectMapper;
+ protected final GeoFormat geoFormat;
+
+ protected JsonNodeToGeometryCodec(
+ TypeCodec targetCodec,
+ ObjectMapper objectMapper,
+ GeoFormat geoFormat,
+ List nullStrings) {
+ super(targetCodec, nullStrings);
+ this.objectMapper = objectMapper;
+ this.geoFormat = geoFormat;
+ }
+
+ @Override
+ public T externalToInternal(JsonNode node) {
+ if (isNullOrEmpty(node)) {
+ return null;
+ }
+ try {
+ // We accept:
+ // 1) String nodes containing WKT literals
+ // 2) String nodes containing Geo JSON documents
+ // 3) String nodes containing WKB encoded strings
+ // 4) Json object nodes compliant with Geo JSON syntax
+ // 5) Binary nodes containing WKB strings
+ String s;
+ if (node.isObject()) {
+ s = objectMapper.writeValueAsString(node);
+ return parseGeometry(s);
+ } else if (node.isBinary()) {
+ byte[] bytes = node.binaryValue();
+ return parseGeometry(bytes);
+ } else {
+ s = node.asText();
+ return parseGeometry(s);
+ }
+ } catch (IOException e) {
+ throw new IllegalArgumentException("Cannot deserialize node " + node, e);
+ }
+ }
+
+ protected abstract T parseGeometry(@NonNull String s);
+
+ protected abstract T parseGeometry(@NonNull byte[] b);
+
+ @Override
+ public JsonNode internalToExternal(T value) {
+ if (value == null) {
+ return null;
+ }
+ try {
+ String s = geoFormat.format(value);
+ if (geoFormat == JsonGeoFormat.INSTANCE) {
+ // Return a JSON object for GeoJson format instead of a text node
+ return objectMapper.readTree(s);
+ } else {
+ // Return a text node for Well-known text and Well-known binary formats.
+ return objectMapper.getNodeFactory().textNode(s);
+ }
+ } catch (IOException e) {
+ throw new IllegalArgumentException("Cannot serialize value " + value, e);
+ }
+ }
+}
diff --git a/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/dse/JsonNodeToLineStringCodec.java b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/dse/JsonNodeToLineStringCodec.java
new file mode 100644
index 0000000..c060230
--- /dev/null
+++ b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/dse/JsonNodeToLineStringCodec.java
@@ -0,0 +1,43 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datastax.oss.dsbulk.codecs.text.json.dse;
+
+import com.datastax.dse.driver.api.core.data.geometry.LineString;
+import com.datastax.dse.driver.api.core.type.codec.DseTypeCodecs;
+import com.datastax.oss.dsbulk.codecs.api.format.geo.GeoFormat;
+import com.datastax.oss.dsbulk.codecs.api.util.CodecUtils;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import edu.umd.cs.findbugs.annotations.NonNull;
+import java.nio.ByteBuffer;
+import java.util.List;
+
+public class JsonNodeToLineStringCodec extends JsonNodeToGeometryCodec {
+
+ public JsonNodeToLineStringCodec(
+ ObjectMapper objectMapper, GeoFormat geoFormat, List nullStrings) {
+ super(DseTypeCodecs.LINE_STRING, objectMapper, geoFormat, nullStrings);
+ }
+
+ @Override
+ protected LineString parseGeometry(@NonNull String s) {
+ return CodecUtils.parseLineString(s);
+ }
+
+ @Override
+ protected LineString parseGeometry(@NonNull byte[] b) {
+ return LineString.fromWellKnownBinary(ByteBuffer.wrap(b));
+ }
+}
diff --git a/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/dse/JsonNodeToPointCodec.java b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/dse/JsonNodeToPointCodec.java
new file mode 100644
index 0000000..1ac02ba
--- /dev/null
+++ b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/dse/JsonNodeToPointCodec.java
@@ -0,0 +1,43 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datastax.oss.dsbulk.codecs.text.json.dse;
+
+import com.datastax.dse.driver.api.core.data.geometry.Point;
+import com.datastax.dse.driver.api.core.type.codec.DseTypeCodecs;
+import com.datastax.oss.dsbulk.codecs.api.format.geo.GeoFormat;
+import com.datastax.oss.dsbulk.codecs.api.util.CodecUtils;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import edu.umd.cs.findbugs.annotations.NonNull;
+import java.nio.ByteBuffer;
+import java.util.List;
+
+public class JsonNodeToPointCodec extends JsonNodeToGeometryCodec {
+
+ public JsonNodeToPointCodec(
+ ObjectMapper objectMapper, GeoFormat geoFormat, List nullStrings) {
+ super(DseTypeCodecs.POINT, objectMapper, geoFormat, nullStrings);
+ }
+
+ @Override
+ protected Point parseGeometry(@NonNull String s) {
+ return CodecUtils.parsePoint(s);
+ }
+
+ @Override
+ protected Point parseGeometry(@NonNull byte[] b) {
+ return Point.fromWellKnownBinary(ByteBuffer.wrap(b));
+ }
+}
diff --git a/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/dse/JsonNodeToPolygonCodec.java b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/dse/JsonNodeToPolygonCodec.java
new file mode 100644
index 0000000..b4fa2b3
--- /dev/null
+++ b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/json/dse/JsonNodeToPolygonCodec.java
@@ -0,0 +1,43 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datastax.oss.dsbulk.codecs.text.json.dse;
+
+import com.datastax.dse.driver.api.core.data.geometry.Polygon;
+import com.datastax.dse.driver.api.core.type.codec.DseTypeCodecs;
+import com.datastax.oss.dsbulk.codecs.api.format.geo.GeoFormat;
+import com.datastax.oss.dsbulk.codecs.api.util.CodecUtils;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import edu.umd.cs.findbugs.annotations.NonNull;
+import java.nio.ByteBuffer;
+import java.util.List;
+
+public class JsonNodeToPolygonCodec extends JsonNodeToGeometryCodec {
+
+ public JsonNodeToPolygonCodec(
+ ObjectMapper objectMapper, GeoFormat geoFormat, List nullStrings) {
+ super(DseTypeCodecs.POLYGON, objectMapper, geoFormat, nullStrings);
+ }
+
+ @Override
+ protected Polygon parseGeometry(@NonNull String s) {
+ return CodecUtils.parsePolygon(s);
+ }
+
+ @Override
+ protected Polygon parseGeometry(@NonNull byte[] b) {
+ return Polygon.fromWellKnownBinary(ByteBuffer.wrap(b));
+ }
+}
diff --git a/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/string/StringConvertingCodec.java b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/string/StringConvertingCodec.java
new file mode 100644
index 0000000..69b0f67
--- /dev/null
+++ b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/string/StringConvertingCodec.java
@@ -0,0 +1,74 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datastax.oss.dsbulk.codecs.text.string;
+
+import com.datastax.oss.driver.api.core.type.codec.TypeCodec;
+import com.datastax.oss.dsbulk.codecs.api.ConvertingCodec;
+import java.util.List;
+
+public abstract class StringConvertingCodec extends ConvertingCodec {
+
+ private final List nullStrings;
+
+ protected StringConvertingCodec(TypeCodec targetCodec, List nullStrings) {
+ super(targetCodec, String.class);
+ this.nullStrings = nullStrings;
+ }
+
+ /**
+ * Whether the input is null.
+ *
+ * This method should be used to inspect external inputs that are meant to be converted to
+ * textual CQL types only (text, varchar and ascii).
+ *
+ *
It always considers the empty string as NOT equivalent to NULL, unless the user clearly
+ * specifies that the empty string is to be considered as NULL, through the
+ * codec.nullStrings setting.
+ *
+ *
Do NOT use this method for non-textual CQL types; use {@link #isNullOrEmpty(String)}
+ * instead.
+ */
+ protected boolean isNull(String s) {
+ return s == null || nullStrings.contains(s);
+ }
+
+ /**
+ * Whether the input is null or empty.
+ *
+ *
This method should be used to inspect external inputs that are meant to be converted to
+ * non-textual CQL types only.
+ *
+ *
It always considers the empty string as equivalent to NULL, which is in compliance with the
+ * documentation of codec.nullStrings: "Note that, regardless of this setting, DSBulk
+ * will always convert empty strings to `null` if the target CQL type is not textual (i.e. not
+ * text, varchar or ascii)."
+ *
+ *
Do NOT use this method for textual CQL types; use {@link #isNull(String)} instead.
+ */
+ protected boolean isNullOrEmpty(String s) {
+ return isNull(s) || s.isEmpty();
+ }
+
+ /**
+ * The string to use when formatting internal inputs.
+ *
+ *
According to the documentation of codec.nullStrings, we must use the first
+ * string specified there, or simply return null, if that setting is empty.
+ */
+ protected String nullString() {
+ return nullStrings.isEmpty() ? null : nullStrings.get(0);
+ }
+}
diff --git a/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/string/StringConvertingCodecProvider.java b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/string/StringConvertingCodecProvider.java
new file mode 100644
index 0000000..3c89984
--- /dev/null
+++ b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/string/StringConvertingCodecProvider.java
@@ -0,0 +1,359 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datastax.oss.dsbulk.codecs.text.string;
+
+import static com.datastax.oss.dsbulk.codecs.api.CommonConversionContext.BINARY_FORMAT;
+import static com.datastax.oss.dsbulk.codecs.api.CommonConversionContext.GEO_FORMAT;
+import static com.datastax.oss.dsbulk.codecs.text.TextConversionContext.BOOLEAN_INPUT_WORDS;
+import static com.datastax.oss.dsbulk.codecs.text.TextConversionContext.BOOLEAN_NUMBERS;
+import static com.datastax.oss.dsbulk.codecs.text.TextConversionContext.BOOLEAN_OUTPUT_WORDS;
+import static com.datastax.oss.dsbulk.codecs.text.TextConversionContext.EPOCH;
+import static com.datastax.oss.dsbulk.codecs.text.TextConversionContext.LOCAL_DATE_FORMAT;
+import static com.datastax.oss.dsbulk.codecs.text.TextConversionContext.LOCAL_TIME_FORMAT;
+import static com.datastax.oss.dsbulk.codecs.text.TextConversionContext.NULL_STRINGS;
+import static com.datastax.oss.dsbulk.codecs.text.TextConversionContext.NUMBER_FORMAT;
+import static com.datastax.oss.dsbulk.codecs.text.TextConversionContext.OBJECT_MAPPER;
+import static com.datastax.oss.dsbulk.codecs.text.TextConversionContext.OVERFLOW_STRATEGY;
+import static com.datastax.oss.dsbulk.codecs.text.TextConversionContext.ROUNDING_MODE;
+import static com.datastax.oss.dsbulk.codecs.text.TextConversionContext.TIMESTAMP_FORMAT;
+import static com.datastax.oss.dsbulk.codecs.text.TextConversionContext.TIME_UNIT;
+import static com.datastax.oss.dsbulk.codecs.text.TextConversionContext.TIME_UUID_GENERATOR;
+import static com.datastax.oss.dsbulk.codecs.text.TextConversionContext.TIME_ZONE;
+import static com.datastax.oss.dsbulk.codecs.text.json.JsonCodecUtils.JSON_NODE_TYPE;
+import static com.datastax.oss.protocol.internal.ProtocolConstants.DataType.ASCII;
+import static com.datastax.oss.protocol.internal.ProtocolConstants.DataType.BIGINT;
+import static com.datastax.oss.protocol.internal.ProtocolConstants.DataType.BLOB;
+import static com.datastax.oss.protocol.internal.ProtocolConstants.DataType.BOOLEAN;
+import static com.datastax.oss.protocol.internal.ProtocolConstants.DataType.COUNTER;
+import static com.datastax.oss.protocol.internal.ProtocolConstants.DataType.CUSTOM;
+import static com.datastax.oss.protocol.internal.ProtocolConstants.DataType.DATE;
+import static com.datastax.oss.protocol.internal.ProtocolConstants.DataType.DECIMAL;
+import static com.datastax.oss.protocol.internal.ProtocolConstants.DataType.DOUBLE;
+import static com.datastax.oss.protocol.internal.ProtocolConstants.DataType.DURATION;
+import static com.datastax.oss.protocol.internal.ProtocolConstants.DataType.FLOAT;
+import static com.datastax.oss.protocol.internal.ProtocolConstants.DataType.INET;
+import static com.datastax.oss.protocol.internal.ProtocolConstants.DataType.INT;
+import static com.datastax.oss.protocol.internal.ProtocolConstants.DataType.LIST;
+import static com.datastax.oss.protocol.internal.ProtocolConstants.DataType.MAP;
+import static com.datastax.oss.protocol.internal.ProtocolConstants.DataType.SET;
+import static com.datastax.oss.protocol.internal.ProtocolConstants.DataType.SMALLINT;
+import static com.datastax.oss.protocol.internal.ProtocolConstants.DataType.TIME;
+import static com.datastax.oss.protocol.internal.ProtocolConstants.DataType.TIMESTAMP;
+import static com.datastax.oss.protocol.internal.ProtocolConstants.DataType.TIMEUUID;
+import static com.datastax.oss.protocol.internal.ProtocolConstants.DataType.TINYINT;
+import static com.datastax.oss.protocol.internal.ProtocolConstants.DataType.TUPLE;
+import static com.datastax.oss.protocol.internal.ProtocolConstants.DataType.UDT;
+import static com.datastax.oss.protocol.internal.ProtocolConstants.DataType.UUID;
+import static com.datastax.oss.protocol.internal.ProtocolConstants.DataType.VARCHAR;
+import static com.datastax.oss.protocol.internal.ProtocolConstants.DataType.VARINT;
+
+import com.datastax.oss.driver.api.core.data.TupleValue;
+import com.datastax.oss.driver.api.core.data.UdtValue;
+import com.datastax.oss.driver.api.core.type.CqlVectorType;
+import com.datastax.oss.driver.api.core.type.CustomType;
+import com.datastax.oss.driver.api.core.type.DataType;
+import com.datastax.oss.driver.api.core.type.DataTypes;
+import com.datastax.oss.driver.api.core.type.codec.CodecNotFoundException;
+import com.datastax.oss.driver.api.core.type.codec.TypeCodec;
+import com.datastax.oss.driver.api.core.type.codec.TypeCodecs;
+import com.datastax.oss.driver.api.core.type.reflect.GenericType;
+import com.datastax.oss.driver.internal.core.type.codec.CqlVectorCodec;
+import com.datastax.oss.driver.internal.core.type.codec.registry.DefaultCodecRegistry;
+import com.datastax.oss.driver.shaded.guava.common.collect.ImmutableList;
+import com.datastax.oss.dsbulk.codecs.api.ConversionContext;
+import com.datastax.oss.dsbulk.codecs.api.ConvertingCodec;
+import com.datastax.oss.dsbulk.codecs.api.ConvertingCodecFactory;
+import com.datastax.oss.dsbulk.codecs.api.ConvertingCodecProvider;
+import com.datastax.oss.dsbulk.codecs.text.string.dse.StringToDateRangeCodec;
+import com.datastax.oss.dsbulk.codecs.text.string.dse.StringToLineStringCodec;
+import com.datastax.oss.dsbulk.codecs.text.string.dse.StringToPointCodec;
+import com.datastax.oss.dsbulk.codecs.text.string.dse.StringToPolygonCodec;
+import com.fasterxml.jackson.databind.JsonNode;
+import edu.umd.cs.findbugs.annotations.NonNull;
+import edu.umd.cs.findbugs.annotations.Nullable;
+import java.time.Instant;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A specialized {@link DefaultCodecRegistry} that is capable of producing {@link ConvertingCodec}s.
+ */
+public class StringConvertingCodecProvider implements ConvertingCodecProvider {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(StringConvertingCodecProvider.class);
+
+ private static final String LINE_STRING_CLASS_NAME =
+ "org.apache.cassandra.db.marshal.LineStringType";
+ private static final String POINT_CLASS_NAME = "org.apache.cassandra.db.marshal.PointType";
+ private static final String POLYGON_CLASS_NAME = "org.apache.cassandra.db.marshal.PolygonType";
+ private static final String DATE_RANGE_CLASS_NAME =
+ "org.apache.cassandra.db.marshal.DateRangeType";
+
+ @NonNull
+ @Override
+ public Optional> maybeProvide(
+ @NonNull DataType cqlType,
+ @NonNull GenericType externalJavaType,
+ @NonNull ConvertingCodecFactory codecFactory,
+ boolean rootCodec) {
+ if (!externalJavaType.equals(GenericType.STRING)) {
+ return Optional.empty();
+ }
+ ConvertingCodec codec =
+ createStringConvertingCodec(cqlType, codecFactory, rootCodec);
+ return Optional.ofNullable(codec);
+ }
+
+ @Nullable
+ private ConvertingCodec createStringConvertingCodec(
+ @NonNull DataType cqlType, @NonNull ConvertingCodecFactory codecFactory, boolean rootCodec) {
+ ConversionContext context = codecFactory.getContext();
+ // DAT-297: Don't apply null strings for non-root codecs
+ List nullStrings = rootCodec ? context.getAttribute(NULL_STRINGS) : ImmutableList.of();
+ int cqlTypeCode = cqlType.getProtocolCode();
+ switch (cqlTypeCode) {
+ case ASCII:
+ case VARCHAR:
+ TypeCodec typeCodec = codecFactory.getCodecRegistry().codecFor(cqlType);
+ return new StringToStringCodec(typeCodec, nullStrings);
+ case BOOLEAN:
+ return new StringToBooleanCodec(
+ context.getAttribute(BOOLEAN_INPUT_WORDS),
+ context.getAttribute(BOOLEAN_OUTPUT_WORDS),
+ nullStrings);
+ case TINYINT:
+ return new StringToByteCodec(
+ context.getAttribute(NUMBER_FORMAT),
+ context.getAttribute(OVERFLOW_STRATEGY),
+ context.getAttribute(ROUNDING_MODE),
+ context.getAttribute(TIMESTAMP_FORMAT),
+ context.getAttribute(TIME_ZONE),
+ context.getAttribute(TIME_UNIT),
+ context.getAttribute(EPOCH),
+ context.getAttribute(BOOLEAN_INPUT_WORDS),
+ context.getAttribute(BOOLEAN_NUMBERS),
+ nullStrings);
+ case SMALLINT:
+ return new StringToShortCodec(
+ context.getAttribute(NUMBER_FORMAT),
+ context.getAttribute(OVERFLOW_STRATEGY),
+ context.getAttribute(ROUNDING_MODE),
+ context.getAttribute(TIMESTAMP_FORMAT),
+ context.getAttribute(TIME_ZONE),
+ context.getAttribute(TIME_UNIT),
+ context.getAttribute(EPOCH),
+ context.getAttribute(BOOLEAN_INPUT_WORDS),
+ context.getAttribute(BOOLEAN_NUMBERS),
+ nullStrings);
+ case INT:
+ return new StringToIntegerCodec(
+ context.getAttribute(NUMBER_FORMAT),
+ context.getAttribute(OVERFLOW_STRATEGY),
+ context.getAttribute(ROUNDING_MODE),
+ context.getAttribute(TIMESTAMP_FORMAT),
+ context.getAttribute(TIME_ZONE),
+ context.getAttribute(TIME_UNIT),
+ context.getAttribute(EPOCH),
+ context.getAttribute(BOOLEAN_INPUT_WORDS),
+ context.getAttribute(BOOLEAN_NUMBERS),
+ nullStrings);
+ case BIGINT:
+ return new StringToLongCodec(
+ TypeCodecs.BIGINT,
+ context.getAttribute(NUMBER_FORMAT),
+ context.getAttribute(OVERFLOW_STRATEGY),
+ context.getAttribute(ROUNDING_MODE),
+ context.getAttribute(TIMESTAMP_FORMAT),
+ context.getAttribute(TIME_ZONE),
+ context.getAttribute(TIME_UNIT),
+ context.getAttribute(EPOCH),
+ context.getAttribute(BOOLEAN_INPUT_WORDS),
+ context.getAttribute(BOOLEAN_NUMBERS),
+ nullStrings);
+ case COUNTER:
+ return new StringToLongCodec(
+ TypeCodecs.COUNTER,
+ context.getAttribute(NUMBER_FORMAT),
+ context.getAttribute(OVERFLOW_STRATEGY),
+ context.getAttribute(ROUNDING_MODE),
+ context.getAttribute(TIMESTAMP_FORMAT),
+ context.getAttribute(TIME_ZONE),
+ context.getAttribute(TIME_UNIT),
+ context.getAttribute(EPOCH),
+ context.getAttribute(BOOLEAN_INPUT_WORDS),
+ context.getAttribute(BOOLEAN_NUMBERS),
+ nullStrings);
+ case FLOAT:
+ return new StringToFloatCodec(
+ context.getAttribute(NUMBER_FORMAT),
+ context.getAttribute(OVERFLOW_STRATEGY),
+ context.getAttribute(ROUNDING_MODE),
+ context.getAttribute(TIMESTAMP_FORMAT),
+ context.getAttribute(TIME_ZONE),
+ context.getAttribute(TIME_UNIT),
+ context.getAttribute(EPOCH),
+ context.getAttribute(BOOLEAN_INPUT_WORDS),
+ context.getAttribute(BOOLEAN_NUMBERS),
+ nullStrings);
+ case DOUBLE:
+ return new StringToDoubleCodec(
+ context.getAttribute(NUMBER_FORMAT),
+ context.getAttribute(OVERFLOW_STRATEGY),
+ context.getAttribute(ROUNDING_MODE),
+ context.getAttribute(TIMESTAMP_FORMAT),
+ context.getAttribute(TIME_ZONE),
+ context.getAttribute(TIME_UNIT),
+ context.getAttribute(EPOCH),
+ context.getAttribute(BOOLEAN_INPUT_WORDS),
+ context.getAttribute(BOOLEAN_NUMBERS),
+ nullStrings);
+ case VARINT:
+ return new StringToBigIntegerCodec(
+ context.getAttribute(NUMBER_FORMAT),
+ context.getAttribute(OVERFLOW_STRATEGY),
+ context.getAttribute(ROUNDING_MODE),
+ context.getAttribute(TIMESTAMP_FORMAT),
+ context.getAttribute(TIME_ZONE),
+ context.getAttribute(TIME_UNIT),
+ context.getAttribute(EPOCH),
+ context.getAttribute(BOOLEAN_INPUT_WORDS),
+ context.getAttribute(BOOLEAN_NUMBERS),
+ nullStrings);
+ case DECIMAL:
+ return new StringToBigDecimalCodec(
+ context.getAttribute(NUMBER_FORMAT),
+ context.getAttribute(OVERFLOW_STRATEGY),
+ context.getAttribute(ROUNDING_MODE),
+ context.getAttribute(TIMESTAMP_FORMAT),
+ context.getAttribute(TIME_ZONE),
+ context.getAttribute(TIME_UNIT),
+ context.getAttribute(EPOCH),
+ context.getAttribute(BOOLEAN_INPUT_WORDS),
+ context.getAttribute(BOOLEAN_NUMBERS),
+ nullStrings);
+ case DATE:
+ return new StringToLocalDateCodec(
+ context.getAttribute(LOCAL_DATE_FORMAT), context.getAttribute(TIME_ZONE), nullStrings);
+ case TIME:
+ return new StringToLocalTimeCodec(
+ context.getAttribute(LOCAL_TIME_FORMAT), context.getAttribute(TIME_ZONE), nullStrings);
+ case TIMESTAMP:
+ return new StringToInstantCodec(
+ context.getAttribute(TIMESTAMP_FORMAT),
+ context.getAttribute(TIME_ZONE),
+ context.getAttribute(EPOCH),
+ nullStrings);
+ case INET:
+ return new StringToInetAddressCodec(nullStrings);
+ case UUID:
+ {
+ ConvertingCodec instantCodec =
+ codecFactory.createConvertingCodec(DataTypes.TIMESTAMP, GenericType.STRING, false);
+ return new StringToUUIDCodec(
+ TypeCodecs.UUID,
+ instantCodec,
+ context.getAttribute(TIME_UUID_GENERATOR),
+ nullStrings);
+ }
+ case TIMEUUID:
+ {
+ ConvertingCodec instantCodec =
+ codecFactory.createConvertingCodec(DataTypes.TIMESTAMP, GenericType.STRING, false);
+ return new StringToUUIDCodec(
+ TypeCodecs.TIMEUUID,
+ instantCodec,
+ context.getAttribute(TIME_UUID_GENERATOR),
+ nullStrings);
+ }
+ case BLOB:
+ return new StringToBlobCodec(nullStrings, context.getAttribute(BINARY_FORMAT));
+ case DURATION:
+ return new StringToDurationCodec(nullStrings);
+ case LIST:
+ {
+ ConvertingCodec> jsonCodec =
+ codecFactory.createConvertingCodec(cqlType, JSON_NODE_TYPE, false);
+ return new StringToListCodec<>(
+ jsonCodec, context.getAttribute(OBJECT_MAPPER), nullStrings);
+ }
+ case SET:
+ {
+ ConvertingCodec> jsonCodec =
+ codecFactory.createConvertingCodec(cqlType, JSON_NODE_TYPE, false);
+ return new StringToSetCodec<>(
+ jsonCodec, context.getAttribute(OBJECT_MAPPER), nullStrings);
+ }
+ case MAP:
+ {
+ ConvertingCodec> jsonCodec =
+ codecFactory.createConvertingCodec(cqlType, JSON_NODE_TYPE, false);
+ return new StringToMapCodec<>(
+ jsonCodec, context.getAttribute(OBJECT_MAPPER), nullStrings);
+ }
+ case TUPLE:
+ {
+ ConvertingCodec jsonCodec =
+ codecFactory.createConvertingCodec(cqlType, JSON_NODE_TYPE, false);
+ return new StringToTupleCodec(
+ jsonCodec, context.getAttribute(OBJECT_MAPPER), nullStrings);
+ }
+ case UDT:
+ {
+ ConvertingCodec jsonCodec =
+ codecFactory.createConvertingCodec(cqlType, JSON_NODE_TYPE, false);
+ return new StringToUDTCodec(jsonCodec, context.getAttribute(OBJECT_MAPPER), nullStrings);
+ }
+ case CUSTOM:
+ {
+ CustomType customType = (CustomType) cqlType;
+ switch (customType.getClassName()) {
+ case POINT_CLASS_NAME:
+ return new StringToPointCodec(context.getAttribute(GEO_FORMAT), nullStrings);
+ case LINE_STRING_CLASS_NAME:
+ return new StringToLineStringCodec(context.getAttribute(GEO_FORMAT), nullStrings);
+ case POLYGON_CLASS_NAME:
+ return new StringToPolygonCodec(context.getAttribute(GEO_FORMAT), nullStrings);
+ case DATE_RANGE_CLASS_NAME:
+ return new StringToDateRangeCodec(nullStrings);
+ case CqlVectorType.CQLVECTOR_CLASS_NAME:
+ CqlVectorType cqlVectorType = (CqlVectorType) cqlType;
+ ConvertingCodec subtypeCodec =
+ codecFactory.createConvertingCodec(
+ cqlVectorType.getSubtype(), GenericType.STRING, false);
+ return new StringToVectorCodec(
+ new CqlVectorCodec(cqlVectorType, subtypeCodec), nullStrings);
+ }
+ }
+ // fall through
+ default:
+ try {
+ TypeCodec innerCodec = codecFactory.getCodecRegistry().codecFor(cqlType);
+ LOGGER.warn(
+ String.format(
+ "CQL type %s is not officially supported by this version of DSBulk; "
+ + "string literals will be parsed and formatted using registered codec %s",
+ cqlType, innerCodec.getClass().getSimpleName()));
+ return new StringToUnknownTypeCodec<>(innerCodec, nullStrings);
+ } catch (CodecNotFoundException ignored) {
+ }
+ return null;
+ }
+ }
+}
diff --git a/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/string/StringToBigDecimalCodec.java b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/string/StringToBigDecimalCodec.java
new file mode 100644
index 0000000..620e497
--- /dev/null
+++ b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/string/StringToBigDecimalCodec.java
@@ -0,0 +1,67 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datastax.oss.dsbulk.codecs.text.string;
+
+import com.datastax.oss.driver.api.core.type.codec.TypeCodecs;
+import com.datastax.oss.dsbulk.codecs.api.format.temporal.TemporalFormat;
+import com.datastax.oss.dsbulk.codecs.api.util.CodecUtils;
+import com.datastax.oss.dsbulk.codecs.api.util.OverflowStrategy;
+import io.netty.util.concurrent.FastThreadLocal;
+import java.math.BigDecimal;
+import java.math.RoundingMode;
+import java.text.NumberFormat;
+import java.time.ZoneId;
+import java.time.ZonedDateTime;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+public class StringToBigDecimalCodec extends StringToNumberCodec {
+
+ public StringToBigDecimalCodec(
+ FastThreadLocal numberFormat,
+ OverflowStrategy overflowStrategy,
+ RoundingMode roundingMode,
+ TemporalFormat temporalFormat,
+ ZoneId timeZone,
+ TimeUnit timeUnit,
+ ZonedDateTime epoch,
+ Map booleanStrings,
+ List booleanNumbers,
+ List nullStrings) {
+ super(
+ TypeCodecs.DECIMAL,
+ numberFormat,
+ overflowStrategy,
+ roundingMode,
+ temporalFormat,
+ timeZone,
+ timeUnit,
+ epoch,
+ booleanStrings,
+ booleanNumbers,
+ nullStrings);
+ }
+
+ @Override
+ public BigDecimal externalToInternal(String s) {
+ Number number = parseNumber(s);
+ if (number == null) {
+ return null;
+ }
+ return CodecUtils.toBigDecimal(number);
+ }
+}
diff --git a/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/string/StringToBigIntegerCodec.java b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/string/StringToBigIntegerCodec.java
new file mode 100644
index 0000000..ee328b4
--- /dev/null
+++ b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/string/StringToBigIntegerCodec.java
@@ -0,0 +1,69 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datastax.oss.dsbulk.codecs.text.string;
+
+import static java.util.stream.Collectors.toList;
+
+import com.datastax.oss.driver.api.core.type.codec.TypeCodecs;
+import com.datastax.oss.dsbulk.codecs.api.format.temporal.TemporalFormat;
+import com.datastax.oss.dsbulk.codecs.api.util.OverflowStrategy;
+import io.netty.util.concurrent.FastThreadLocal;
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.math.RoundingMode;
+import java.text.NumberFormat;
+import java.time.ZoneId;
+import java.time.ZonedDateTime;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+public class StringToBigIntegerCodec extends StringToNumberCodec {
+
+ public StringToBigIntegerCodec(
+ FastThreadLocal numberFormat,
+ OverflowStrategy overflowStrategy,
+ RoundingMode roundingMode,
+ TemporalFormat temporalFormat,
+ ZoneId timeZone,
+ TimeUnit timeUnit,
+ ZonedDateTime epoch,
+ Map booleanStrings,
+ List booleanNumbers,
+ List nullStrings) {
+ super(
+ TypeCodecs.VARINT,
+ numberFormat,
+ overflowStrategy,
+ roundingMode,
+ temporalFormat,
+ timeZone,
+ timeUnit,
+ epoch,
+ booleanStrings,
+ booleanNumbers.stream().map(BigDecimal::toBigInteger).collect(toList()),
+ nullStrings);
+ }
+
+ @Override
+ public BigInteger externalToInternal(String s) {
+ Number number = parseNumber(s);
+ if (number == null) {
+ return null;
+ }
+ return narrowNumber(number, BigInteger.class);
+ }
+}
diff --git a/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/string/StringToBlobCodec.java b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/string/StringToBlobCodec.java
new file mode 100644
index 0000000..50bc21a
--- /dev/null
+++ b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/string/StringToBlobCodec.java
@@ -0,0 +1,49 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datastax.oss.dsbulk.codecs.text.string;
+
+import com.datastax.oss.driver.api.core.type.codec.TypeCodecs;
+import com.datastax.oss.dsbulk.codecs.api.format.binary.BinaryFormat;
+import com.datastax.oss.dsbulk.codecs.api.util.CodecUtils;
+import java.nio.ByteBuffer;
+import java.util.List;
+
+public class StringToBlobCodec extends StringConvertingCodec {
+
+ private final BinaryFormat binaryFormat;
+
+ public StringToBlobCodec(List nullStrings, BinaryFormat binaryFormat) {
+ super(TypeCodecs.BLOB, nullStrings);
+ this.binaryFormat = binaryFormat;
+ }
+
+ @Override
+ public ByteBuffer externalToInternal(String s) {
+ // DAT-573: consider empty strings as empty byte arrays
+ if (isNull(s)) {
+ return null;
+ }
+ return CodecUtils.parseByteBuffer(s);
+ }
+
+ @Override
+ public String internalToExternal(ByteBuffer value) {
+ if (value == null) {
+ return nullString();
+ }
+ return binaryFormat.format(value);
+ }
+}
diff --git a/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/string/StringToBooleanCodec.java b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/string/StringToBooleanCodec.java
new file mode 100644
index 0000000..198b2e7
--- /dev/null
+++ b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/string/StringToBooleanCodec.java
@@ -0,0 +1,57 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datastax.oss.dsbulk.codecs.text.string;
+
+import com.datastax.oss.driver.api.core.type.codec.TypeCodecs;
+import java.util.List;
+import java.util.Map;
+
+public class StringToBooleanCodec extends StringConvertingCodec {
+
+ private final Map inputs;
+ private final Map outputs;
+
+ public StringToBooleanCodec(
+ Map inputs, Map outputs, List nullStrings) {
+ super(TypeCodecs.BOOLEAN, nullStrings);
+ this.inputs = inputs;
+ this.outputs = outputs;
+ }
+
+ @Override
+ public Boolean externalToInternal(String s) {
+ if (isNullOrEmpty(s)) {
+ return null;
+ }
+ Boolean b = inputs.get(s.toLowerCase());
+ if (b == null) {
+ throw new IllegalArgumentException("Invalid boolean value: " + s);
+ }
+ return b;
+ }
+
+ @Override
+ public String internalToExternal(Boolean value) {
+ if (value == null) {
+ return nullString();
+ }
+ String s = outputs.get(value);
+ if (s == null) {
+ return Boolean.toString(value);
+ }
+ return s;
+ }
+}
diff --git a/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/string/StringToByteCodec.java b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/string/StringToByteCodec.java
new file mode 100644
index 0000000..e123adf
--- /dev/null
+++ b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/string/StringToByteCodec.java
@@ -0,0 +1,68 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datastax.oss.dsbulk.codecs.text.string;
+
+import static java.util.stream.Collectors.toList;
+
+import com.datastax.oss.driver.api.core.type.codec.TypeCodecs;
+import com.datastax.oss.dsbulk.codecs.api.format.temporal.TemporalFormat;
+import com.datastax.oss.dsbulk.codecs.api.util.OverflowStrategy;
+import io.netty.util.concurrent.FastThreadLocal;
+import java.math.BigDecimal;
+import java.math.RoundingMode;
+import java.text.NumberFormat;
+import java.time.ZoneId;
+import java.time.ZonedDateTime;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+public class StringToByteCodec extends StringToNumberCodec {
+
+ public StringToByteCodec(
+ FastThreadLocal numberFormat,
+ OverflowStrategy overflowStrategy,
+ RoundingMode roundingMode,
+ TemporalFormat temporalFormat,
+ ZoneId timeZone,
+ TimeUnit timeUnit,
+ ZonedDateTime epoch,
+ Map booleanStrings,
+ List booleanNumbers,
+ List nullStrings) {
+ super(
+ TypeCodecs.TINYINT,
+ numberFormat,
+ overflowStrategy,
+ roundingMode,
+ temporalFormat,
+ timeZone,
+ timeUnit,
+ epoch,
+ booleanStrings,
+ booleanNumbers.stream().map(BigDecimal::byteValueExact).collect(toList()),
+ nullStrings);
+ }
+
+ @Override
+ public Byte externalToInternal(String s) {
+ Number number = parseNumber(s);
+ if (number == null) {
+ return null;
+ }
+ return narrowNumber(number, Byte.class);
+ }
+}
diff --git a/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/string/StringToCollectionCodec.java b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/string/StringToCollectionCodec.java
new file mode 100644
index 0000000..bf11c50
--- /dev/null
+++ b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/string/StringToCollectionCodec.java
@@ -0,0 +1,66 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datastax.oss.dsbulk.codecs.text.string;
+
+import com.datastax.oss.dsbulk.codecs.api.ConvertingCodec;
+import com.datastax.oss.dsbulk.codecs.text.utils.StringUtils;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.io.IOException;
+import java.util.Collection;
+import java.util.List;
+
+public abstract class StringToCollectionCodec>
+ extends StringConvertingCodec {
+
+ private final ConvertingCodec jsonCodec;
+ private final ObjectMapper objectMapper;
+
+ StringToCollectionCodec(
+ ConvertingCodec jsonCodec, ObjectMapper objectMapper, List nullStrings) {
+ super(jsonCodec.getInternalCodec(), nullStrings);
+ this.jsonCodec = jsonCodec;
+ this.objectMapper = objectMapper;
+ }
+
+ @Override
+ public C externalToInternal(String s) {
+ if (isNullOrEmpty(s)) {
+ return null;
+ }
+ try {
+ JsonNode node = objectMapper.readTree(StringUtils.ensureBrackets(s));
+ return jsonCodec.externalToInternal(node);
+ } catch (IOException e) {
+ throw new IllegalArgumentException(String.format("Could not parse '%s' as Json", s), e);
+ }
+ }
+
+ @Override
+ public String internalToExternal(C collection) {
+ if (collection == null) {
+ return nullString();
+ }
+ try {
+ JsonNode node = jsonCodec.internalToExternal(collection);
+ return objectMapper.writeValueAsString(node);
+ } catch (JsonProcessingException e) {
+ throw new IllegalArgumentException(
+ String.format("Could not format '%s' to Json", collection), e);
+ }
+ }
+}
diff --git a/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/string/StringToDoubleCodec.java b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/string/StringToDoubleCodec.java
new file mode 100644
index 0000000..c8e5254
--- /dev/null
+++ b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/string/StringToDoubleCodec.java
@@ -0,0 +1,68 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datastax.oss.dsbulk.codecs.text.string;
+
+import static java.util.stream.Collectors.toList;
+
+import com.datastax.oss.driver.api.core.type.codec.TypeCodecs;
+import com.datastax.oss.dsbulk.codecs.api.format.temporal.TemporalFormat;
+import com.datastax.oss.dsbulk.codecs.api.util.OverflowStrategy;
+import io.netty.util.concurrent.FastThreadLocal;
+import java.math.BigDecimal;
+import java.math.RoundingMode;
+import java.text.NumberFormat;
+import java.time.ZoneId;
+import java.time.ZonedDateTime;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+public class StringToDoubleCodec extends StringToNumberCodec {
+
+ public StringToDoubleCodec(
+ FastThreadLocal numberFormat,
+ OverflowStrategy overflowStrategy,
+ RoundingMode roundingMode,
+ TemporalFormat temporalFormat,
+ ZoneId timeZone,
+ TimeUnit timeUnit,
+ ZonedDateTime epoch,
+ Map booleanStrings,
+ List booleanNumbers,
+ List nullStrings) {
+ super(
+ TypeCodecs.DOUBLE,
+ numberFormat,
+ overflowStrategy,
+ roundingMode,
+ temporalFormat,
+ timeZone,
+ timeUnit,
+ epoch,
+ booleanStrings,
+ booleanNumbers.stream().map(BigDecimal::doubleValue).collect(toList()),
+ nullStrings);
+ }
+
+ @Override
+ public Double externalToInternal(String s) {
+ Number number = parseNumber(s);
+ if (number == null) {
+ return null;
+ }
+ return narrowNumber(number, Double.class);
+ }
+}
diff --git a/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/string/StringToDurationCodec.java b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/string/StringToDurationCodec.java
new file mode 100644
index 0000000..411c2e1
--- /dev/null
+++ b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/string/StringToDurationCodec.java
@@ -0,0 +1,43 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datastax.oss.dsbulk.codecs.text.string;
+
+import com.datastax.oss.driver.api.core.data.CqlDuration;
+import com.datastax.oss.driver.api.core.type.codec.TypeCodecs;
+import java.util.List;
+
+public class StringToDurationCodec extends StringConvertingCodec {
+
+ public StringToDurationCodec(List nullStrings) {
+ super(TypeCodecs.DURATION, nullStrings);
+ }
+
+ @Override
+ public CqlDuration externalToInternal(String s) {
+ if (isNullOrEmpty(s)) {
+ return null;
+ }
+ return CqlDuration.from(s);
+ }
+
+ @Override
+ public String internalToExternal(CqlDuration value) {
+ if (value == null) {
+ return nullString();
+ }
+ return value.toString();
+ }
+}
diff --git a/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/string/StringToFloatCodec.java b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/string/StringToFloatCodec.java
new file mode 100644
index 0000000..ad46e42
--- /dev/null
+++ b/text/src/main/java/com/datastax/oss/dsbulk/codecs/text/string/StringToFloatCodec.java
@@ -0,0 +1,68 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datastax.oss.dsbulk.codecs.text.string;
+
+import static java.util.stream.Collectors.toList;
+
+import com.datastax.oss.driver.api.core.type.codec.TypeCodecs;
+import com.datastax.oss.dsbulk.codecs.api.format.temporal.TemporalFormat;
+import com.datastax.oss.dsbulk.codecs.api.util.OverflowStrategy;
+import io.netty.util.concurrent.FastThreadLocal;
+import java.math.BigDecimal;
+import java.math.RoundingMode;
+import java.text.NumberFormat;
+import java.time.ZoneId;
+import java.time.ZonedDateTime;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+public class StringToFloatCodec extends StringToNumberCodec {
+
+ public StringToFloatCodec(
+ FastThreadLocal numberFormat,
+ OverflowStrategy overflowStrategy,
+ RoundingMode roundingMode,
+ TemporalFormat temporalFormat,
+ ZoneId timeZone,
+ TimeUnit timeUnit,
+ ZonedDateTime epoch,
+ Map