>From Hussain Towaileb <[email protected]>:
Hussain Towaileb has uploaded this change for review. (
https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/21391?usp=email )
Change subject: Add support to Variant type
......................................................................
Add support to Variant type
Change-Id: Iaf8cbd6657cabcf15008501b53892a1432d873eb
---
M
asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/parser/IcebergParquetDataParser.java
1 file changed, 178 insertions(+), 35 deletions(-)
git pull ssh://asterix-gerrit.ics.uci.edu:29418/asterixdb
refs/changes/91/21391/1
diff --git
a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/parser/IcebergParquetDataParser.java
b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/parser/IcebergParquetDataParser.java
index 7ed1b24..665e464 100644
---
a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/parser/IcebergParquetDataParser.java
+++
b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/parser/IcebergParquetDataParser.java
@@ -36,7 +36,6 @@
import java.time.zone.ZoneRules;
import java.util.List;
import java.util.Map;
-import java.util.UUID;
import java.util.concurrent.TimeUnit;
import org.apache.asterix.builders.IARecordBuilder;
@@ -59,6 +58,7 @@
import org.apache.hyracks.api.exceptions.Warning;
import org.apache.hyracks.data.std.api.IMutableValueStorage;
import org.apache.hyracks.data.std.api.IValueReference;
+import org.apache.hyracks.util.annotations.AiProvenance;
import org.apache.iceberg.Schema;
import org.apache.iceberg.StructLike;
import org.apache.iceberg.data.Record;
@@ -66,6 +66,11 @@
import org.apache.iceberg.types.Types;
import org.apache.iceberg.types.Types.NestedField;
import org.apache.iceberg.types.Types.StructType;
+import org.apache.iceberg.variants.PhysicalType;
+import org.apache.iceberg.variants.Variant;
+import org.apache.iceberg.variants.VariantArray;
+import org.apache.iceberg.variants.VariantObject;
+import org.apache.iceberg.variants.VariantValue;
public class IcebergParquetDataParser extends AbstractDataParser implements
IRecordDataParser<Record> {
private final IcebergConverterContext parserContext;
@@ -126,15 +131,16 @@
parserContext.exitObject(valueBuffer, null, objectBuilder);
}
+ @AiProvenance(agent = AiProvenance.Agent.CLAUDE_SONNET_4_6, tool =
AiProvenance.Tool.CLAUDE_CODE_UI, contributionKind =
AiProvenance.ContributionKind.REFACTORED, notes = "Added VARIANT case")
private void parseValue(Type fieldType, Object value, DataOutput out)
throws IOException {
if (value == null) {
- nullSerde.serialize(ANull.NULL, out);
+ serializeNull(out);
return;
}
switch (fieldType.typeId()) {
case BOOLEAN:
- booleanSerde.serialize((Boolean) value ? ABoolean.TRUE :
ABoolean.FALSE, out);
+ serializeBoolean((Boolean) value, out);
return;
case INTEGER:
serializeInteger(value, out);
@@ -152,7 +158,7 @@
serializeString(value, out);
return;
case UUID:
- serializeUuid(value, out);
+ serializeUuid(value.toString(), out);
return;
case FIXED:
serializeFixedBinary(value, out);
@@ -185,9 +191,11 @@
case TIMESTAMP_NANO:
serializeTimestamp(fieldType, value, out);
return;
+ case VARIANT:
+ parseVariant((Variant) value, out);
+ return;
case GEOMETRY:
case GEOGRAPHY:
- case VARIANT:
case UNKNOWN:
default:
throw createUnsupportedException(fieldType);
@@ -197,10 +205,9 @@
private void parseArray(Types.ListType listType, List<?> listValues,
DataOutput out) throws IOException {
if (listValues == null) {
- nullSerde.serialize(ANull.NULL, out);
+ serializeNull(out);
return;
}
-
Type elementType = listType.elementType();
final IMutableValueStorage valueBuffer =
parserContext.enterCollection();
final IAsterixListBuilder arrayBuilder =
parserContext.getCollectionBuilder(NESTED_OPEN_AORDERED_LIST_TYPE);
@@ -259,6 +266,101 @@
parserContext.exitCollection(item, listBuilder);
}
+ @AiProvenance(agent = AiProvenance.Agent.CLAUDE_SONNET_4_6, tool =
AiProvenance.Tool.CLAUDE_CODE_UI, contributionKind =
AiProvenance.ContributionKind.GENERATED, notes = "Entry point for VARIANT
column: unwraps Iceberg Variant and delegates to parseVariantValue")
+ private void parseVariant(Variant variant, DataOutput out) throws
IOException {
+ parseVariantValue(variant.value(), out);
+ }
+
+ @AiProvenance(agent = AiProvenance.Agent.CLAUDE_SONNET_4_6, tool =
AiProvenance.Tool.CLAUDE_CODE_UI, contributionKind =
AiProvenance.ContributionKind.GENERATED, notes = "Dispatches on PhysicalType
and serializes each Variant primitive/object/array to the proper type")
+ private void parseVariantValue(VariantValue variantValue, DataOutput out)
throws IOException {
+ PhysicalType physicalType = variantValue.type();
+ switch (physicalType) {
+ case NULL:
+ serializeNull(out);
+ return;
+ case BOOLEAN_TRUE:
+ serializeBoolean(true, out);
+ return;
+ case BOOLEAN_FALSE:
+ serializeBoolean(false, out);
+ return;
+ case INT8:
+ case INT16:
+ case INT32:
+ case INT64:
+ serializeLong(((Number)
variantValue.asPrimitive().get()).longValue(), out);
+ return;
+ case FLOAT:
+ serializeFloat(variantValue.asPrimitive().get(), out);
+ return;
+ case DOUBLE:
+ serializeDouble(variantValue.asPrimitive().get(), out);
+ return;
+ case DECIMAL4:
+ case DECIMAL8:
+ case DECIMAL16:
+ serializeDecimal((BigDecimal)
variantValue.asPrimitive().get(), out);
+ return;
+ case STRING:
+ serializeString(variantValue.asPrimitive().get(), out);
+ return;
+ case BINARY:
+ serializeBinary(variantValue.asPrimitive().get(), out);
+ return;
+ case DATE:
+ serializeDateEpochDay((Integer)
variantValue.asPrimitive().get(), out);
+ return;
+ case TIME:
+ serializeTimeMicros((Long) variantValue.asPrimitive().get(),
out);
+ return;
+ case TIMESTAMPTZ:
+ case TIMESTAMPNTZ:
+ serializeTimestampMicros((Long)
variantValue.asPrimitive().get(), out);
+ return;
+ case TIMESTAMPTZ_NANOS:
+ case TIMESTAMPNTZ_NANOS:
+ serializeTimestampNanos((Long)
variantValue.asPrimitive().get(), out);
+ return;
+ case UUID:
+ serializeUuid((String) variantValue.asPrimitive().get(), out);
+ return;
+ case OBJECT:
+ parseVariantObject(variantValue.asObject(), out);
+ return;
+ case ARRAY:
+ parseVariantArray(variantValue.asArray(), out);
+ return;
+ default:
+ throw createVariantUnsupportedException(physicalType);
+ }
+ }
+
+ @AiProvenance(agent = AiProvenance.Agent.CLAUDE_SONNET_4_6, tool =
AiProvenance.Tool.CLAUDE_CODE_UI, contributionKind =
AiProvenance.ContributionKind.GENERATED, notes = "Serializes a VariantObject as
an open AsterixDB record")
+ private void parseVariantObject(VariantObject variantObject, DataOutput
out) throws IOException {
+ IMutableValueStorage valueBuffer = parserContext.enterObject();
+ IARecordBuilder objectBuilder =
parserContext.getObjectBuilder(DefaultOpenFieldType.NESTED_OPEN_RECORD_TYPE);
+ for (String fieldName : variantObject.fieldNames()) {
+ valueBuffer.reset();
+ parseVariantValue(variantObject.get(fieldName),
valueBuffer.getDataOutput());
+
objectBuilder.addField(parserContext.getSerializedFieldName(fieldName),
valueBuffer);
+ }
+ objectBuilder.write(out, true);
+ parserContext.exitObject(valueBuffer, null, objectBuilder);
+ }
+
+ @AiProvenance(agent = AiProvenance.Agent.CLAUDE_SONNET_4_6, tool =
AiProvenance.Tool.CLAUDE_CODE_UI, contributionKind =
AiProvenance.ContributionKind.GENERATED, notes = "Serializes a VariantArray as
an AsterixDB ordered list")
+ private void parseVariantArray(VariantArray variantArray, DataOutput out)
throws IOException {
+ final IMutableValueStorage valueBuffer =
parserContext.enterCollection();
+ final IAsterixListBuilder arrayBuilder =
parserContext.getCollectionBuilder(NESTED_OPEN_AORDERED_LIST_TYPE);
+ for (int i = 0; i < variantArray.numElements(); i++) {
+ valueBuffer.reset();
+ parseVariantValue(variantArray.get(i),
valueBuffer.getDataOutput());
+ arrayBuilder.addItem(valueBuffer);
+ }
+ arrayBuilder.write(out, true);
+ parserContext.exitCollection(valueBuffer, arrayBuilder);
+ }
+
private void parseValueAndAddObjectField(IMutableValueStorage valueBuffer,
IARecordBuilder objectBuilder,
Type valueType, String fieldName, Object fieldValue) throws
IOException {
ATypeTag typeTag = getTypeTag(valueType, fieldValue == null,
parserContext);
@@ -277,6 +379,14 @@
}
}
+ private void serializeNull(DataOutput out) throws HyracksDataException {
+ nullSerde.serialize(ANull.NULL, out);
+ }
+
+ private void serializeBoolean(boolean value, DataOutput out) throws
HyracksDataException {
+ booleanSerde.serialize(value ? ABoolean.TRUE : ABoolean.FALSE, out);
+ }
+
private void serializeInteger(Object value, DataOutput out) throws
HyracksDataException {
int intValue = (Integer) value;
aInt64.setValue(intValue);
@@ -301,11 +411,8 @@
doubleSerde.serialize(aDouble, out);
}
- private void serializeUuid(Object value, DataOutput out) throws
HyracksDataException {
- UUID uuid = (UUID) value;
- String uuidValue = uuid.toString();
- char[] buffer = uuidValue.toCharArray();
- aUUID.parseUUIDString(buffer, 0, uuidValue.length());
+ private void serializeUuid(String uuidStr, DataOutput out) throws
HyracksDataException {
+ aUUID.parseUUIDString(uuidStr.toCharArray(), 0, uuidStr.length());
uuidSerde.serialize(aUUID, out);
}
@@ -344,26 +451,38 @@
}
private void serializeDate(Object value, DataOutput output) throws
HyracksDataException {
- LocalDate localDate = (LocalDate) value;
+ serializeDateEpochDay((int) ((LocalDate) value).toEpochDay(), output);
+ }
+
+ @AiProvenance(agent = AiProvenance.Agent.CLAUDE_SONNET_4_6, tool =
AiProvenance.Tool.CLAUDE_CODE_UI, contributionKind =
AiProvenance.ContributionKind.GENERATED, notes = "Core date serializer
accepting days-since-epoch; shared by serializeDate and Variant DATE handling")
+ private void serializeDateEpochDay(int epochDay, DataOutput out) throws
HyracksDataException {
if (parserContext.isDateAsInt()) {
- serializeInteger((int) localDate.toEpochDay(), output);
+ serializeInteger(epochDay, out);
} else {
- aDate.setValue((int) localDate.toEpochDay());
- dateSerde.serialize(aDate, output);
+ aDate.setValue(epochDay);
+ dateSerde.serialize(aDate, out);
}
}
private void serializeTime(Object value, DataOutput output) throws
HyracksDataException {
- LocalTime localTime = (LocalTime) value;
- int timeInMillis = (int)
TimeUnit.NANOSECONDS.toMillis(localTime.toNanoOfDay());
+ serializeTimeMillis((int) TimeUnit.NANOSECONDS.toMillis(((LocalTime)
value).toNanoOfDay()), output);
+ }
+
+ @AiProvenance(agent = AiProvenance.Agent.CLAUDE_SONNET_4_6, tool =
AiProvenance.Tool.CLAUDE_CODE_UI, contributionKind =
AiProvenance.ContributionKind.GENERATED, notes = "Core time serializer
accepting milliseconds-of-day; shared by serializeTime and serializeTimeMicros")
+ private void serializeTimeMillis(int timeInMillis, DataOutput out) throws
HyracksDataException {
if (parserContext.isTimeAsInt()) {
- serializeInteger(timeInMillis, output);
+ serializeInteger(timeInMillis, out);
} else {
aTime.setValue(timeInMillis);
- timeSerde.serialize(aTime, output);
+ timeSerde.serialize(aTime, out);
}
}
+ @AiProvenance(agent = AiProvenance.Agent.CLAUDE_SONNET_4_6, tool =
AiProvenance.Tool.CLAUDE_CODE_UI, contributionKind =
AiProvenance.ContributionKind.GENERATED, notes = "Converts Variant TIME
(microseconds-of-day) to milliseconds and delegates to serializeTimeMillis")
+ private void serializeTimeMicros(long timeMicros, DataOutput out) throws
HyracksDataException {
+ serializeTimeMillis((int) TimeUnit.MICROSECONDS.toMillis(timeMicros),
out);
+ }
+
private void serializeTimestamp(Type type, Object value, DataOutput
output) throws HyracksDataException {
switch (value) {
case LocalDateTime localDateTime ->
serializeWallClockTimestamp(type, localDateTime, output);
@@ -378,12 +497,10 @@
throws HyracksDataException {
long epochSecond = localDateTime.toEpochSecond(ZoneOffset.UTC);
int nano = localDateTime.getNano();
- if (parserContext.isTimestampAsLong()) {
- long value = isTimestampNano(type) ? toNanos(epochSecond, nano) :
toMicros(epochSecond, nano);
- serializeLong(value, output);
+ if (isTimestampNano(type)) {
+ serializeTimestampNanos(toNanos(epochSecond, nano), output);
} else {
- aDateTime.setValue(toMillis(epochSecond, nano));
- datetimeSerde.serialize(aDateTime, output);
+ serializeTimestampMicros(toMicros(epochSecond, nano), output);
}
}
@@ -391,19 +508,40 @@
throws HyracksDataException {
long epochSecond = offsetDateTime.toEpochSecond();
int nano = offsetDateTime.getNano();
- if (parserContext.isTimestampAsLong()) {
- TimestampUnit unit = getTimestampUnit(type);
- long value = unit == TimestampUnit.NANOS ? toNanos(epochSecond,
nano) : toMicros(epochSecond, nano);
- value = timestampZoneProjector.projectEpochValue(value, unit);
- serializeLong(value, output);
+ if (isTimestampNano(type)) {
+ serializeTimestampNanos(
+
timestampZoneProjector.projectEpochValue(toNanos(epochSecond, nano),
TimestampUnit.NANOS), output);
} else {
- long epochMillis = toMillis(epochSecond, nano);
- epochMillis =
timestampZoneProjector.projectEpochMillis(epochMillis);
- aDateTime.setValue(epochMillis);
- datetimeSerde.serialize(aDateTime, output);
+ serializeTimestampMicros(
+
timestampZoneProjector.projectEpochValue(toMicros(epochSecond, nano),
TimestampUnit.MICROS),
+ output);
}
}
+ @AiProvenance(agent = AiProvenance.Agent.CLAUDE_SONNET_4_6, tool =
AiProvenance.Tool.CLAUDE_CODE_UI, contributionKind =
AiProvenance.ContributionKind.GENERATED, notes = "Handles isTimestampAsLong
flag for microsecond-precision timestamps; shared by Iceberg TIMESTAMP and
Variant TIMESTAMPTZ/TIMESTAMPNTZ")
+ private void serializeTimestampMicros(long epochMicros, DataOutput out)
throws HyracksDataException {
+ if (parserContext.isTimestampAsLong()) {
+ serializeLong(epochMicros, out);
+ } else {
+
serializeDatetimeMillis(TimeUnit.MICROSECONDS.toMillis(epochMicros), out);
+ }
+ }
+
+ @AiProvenance(agent = AiProvenance.Agent.CLAUDE_SONNET_4_6, tool =
AiProvenance.Tool.CLAUDE_CODE_UI, contributionKind =
AiProvenance.ContributionKind.GENERATED, notes = "Handles isTimestampAsLong
flag for nanosecond-precision timestamps; shared by Iceberg TIMESTAMP_NANO and
Variant TIMESTAMPTZ_NANOS/TIMESTAMPNTZ_NANOS")
+ private void serializeTimestampNanos(long epochNanos, DataOutput out)
throws HyracksDataException {
+ if (parserContext.isTimestampAsLong()) {
+ serializeLong(epochNanos, out);
+ } else {
+ serializeDatetimeMillis(TimeUnit.NANOSECONDS.toMillis(epochNanos),
out);
+ }
+ }
+
+ @AiProvenance(agent = AiProvenance.Agent.CLAUDE_SONNET_4_6, tool =
AiProvenance.Tool.CLAUDE_CODE_UI, contributionKind =
AiProvenance.ContributionKind.GENERATED, notes = "Core datetime serializer
accepting epoch milliseconds; shared by timestamp helpers and Variant TIMESTAMP
handling")
+ private void serializeDatetimeMillis(long epochMillis, DataOutput out)
throws HyracksDataException {
+ aDateTime.setValue(epochMillis);
+ datetimeSerde.serialize(aDateTime, out);
+ }
+
private static TimestampUnit getTimestampUnit(Type type) {
return isTimestampNano(type) ? TimestampUnit.NANOS :
TimestampUnit.MICROS;
}
@@ -444,6 +582,10 @@
return new RuntimeDataException(ErrorCode.TYPE_UNSUPPORTED, "Iceberg
Parser", type.toString());
}
+ private static HyracksDataException
createVariantUnsupportedException(PhysicalType type) {
+ return new RuntimeDataException(ErrorCode.TYPE_UNSUPPORTED, "Iceberg
Variant Parser", type.toString());
+ }
+
private static void ensureDecimalToDoubleEnabled(Type type,
IcebergConverterContext context)
throws RuntimeDataException {
if (!context.isDecimalToDoubleEnabled()) {
@@ -452,6 +594,7 @@
}
}
+ @AiProvenance(agent = AiProvenance.Agent.CLAUDE_SONNET_4_6, tool =
AiProvenance.Tool.CLAUDE_CODE_UI, contributionKind =
AiProvenance.ContributionKind.REFACTORED, notes = "Added VARIANT case")
private static ATypeTag getTypeTag(Type type, boolean isNull,
IcebergConverterContext parserContext)
throws HyracksDataException {
if (isNull) {
@@ -470,7 +613,7 @@
ensureDecimalToDoubleEnabled(type, parserContext);
yield ATypeTag.DOUBLE;
}
- case STRUCT -> ATypeTag.OBJECT;
+ case STRUCT, VARIANT -> ATypeTag.OBJECT;
case LIST, MAP -> ATypeTag.ARRAY;
case DATE -> {
if (parserContext.isDateAsInt()) {
--
To view, visit https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/21391?usp=email
To unsubscribe, or for help writing mail filters, visit
https://asterix-gerrit.ics.uci.edu/settings?usp=email
Gerrit-MessageType: newchange
Gerrit-Project: asterixdb
Gerrit-Branch: master
Gerrit-Change-Id: Iaf8cbd6657cabcf15008501b53892a1432d873eb
Gerrit-Change-Number: 21391
Gerrit-PatchSet: 1
Gerrit-Owner: Hussain Towaileb <[email protected]>
