pvary commented on code in PR #17024:
URL: https://github.com/apache/iceberg/pull/17024#discussion_r3537542200
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/CompareSchemasVisitor.java:
##########
@@ -180,27 +179,43 @@ public Result primitive(Type.PrimitiveType primitive,
Integer tableSchemaId) {
Type.PrimitiveType tableSchemaPrimitiveType =
tableSchemaType.asPrimitiveType();
if (primitive.equals(tableSchemaPrimitiveType)) {
return Result.SAME;
- } else if (primitive.equals(Types.IntegerType.get())
- && tableSchemaPrimitiveType.equals(Types.LongType.get())) {
- return Result.DATA_CONVERSION_NEEDED;
- } else if (primitive.equals(Types.FloatType.get())
- && tableSchemaPrimitiveType.equals(Types.DoubleType.get())) {
+ } else if (isDataConversionPossible(primitive, tableSchemaPrimitiveType)) {
return Result.DATA_CONVERSION_NEEDED;
- } else if (primitive.equals(Types.DateType.get())
- && tableSchemaPrimitiveType.equals(Types.TimestampType.withoutZone()))
{
- return Result.DATA_CONVERSION_NEEDED;
- } else if (primitive.typeId() == Type.TypeID.DECIMAL
- && tableSchemaPrimitiveType.typeId() == Type.TypeID.DECIMAL) {
- Types.DecimalType dataType = (Types.DecimalType) primitive;
- Types.DecimalType tableType = (Types.DecimalType)
tableSchemaPrimitiveType;
- return dataType.scale() == tableType.scale() && dataType.precision() <
tableType.precision()
- ? Result.DATA_CONVERSION_NEEDED
- : Result.SCHEMA_UPDATE_NEEDED;
} else {
return Result.SCHEMA_UPDATE_NEEDED;
}
}
+ /**
+ * Whether {@link DataConverter} can convert input data of type {@code
dataType} into the table's
+ * {@code tableType}. Must stay in sync with the conversions {@link
DataConverter#get} performs.
+ */
+ @SuppressWarnings("checkstyle:CyclomaticComplexity")
+ static boolean isDataConversionPossible(
+ Type.PrimitiveType dataType, Type.PrimitiveType tableType) {
+ if (dataType.equals(Types.IntegerType.get()) &&
tableType.equals(Types.LongType.get())) {
+ return true;
+ }
+
+ if (dataType.equals(Types.FloatType.get()) &&
tableType.equals(Types.DoubleType.get())) {
+ return true;
+ }
+
+ if (dataType.equals(Types.DateType.get())
+ && tableType.equals(Types.TimestampType.withoutZone())) {
+ return true;
+ }
Review Comment:
I'm ok fixing/testing the dynsink with nano in a different PR.
Spec propotion rules are a bigger task, as they might have requirements
outside Flink
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/CompareSchemasVisitor.java:
##########
@@ -180,27 +179,43 @@ public Result primitive(Type.PrimitiveType primitive,
Integer tableSchemaId) {
Type.PrimitiveType tableSchemaPrimitiveType =
tableSchemaType.asPrimitiveType();
if (primitive.equals(tableSchemaPrimitiveType)) {
return Result.SAME;
- } else if (primitive.equals(Types.IntegerType.get())
- && tableSchemaPrimitiveType.equals(Types.LongType.get())) {
- return Result.DATA_CONVERSION_NEEDED;
- } else if (primitive.equals(Types.FloatType.get())
- && tableSchemaPrimitiveType.equals(Types.DoubleType.get())) {
+ } else if (isDataConversionPossible(primitive, tableSchemaPrimitiveType)) {
return Result.DATA_CONVERSION_NEEDED;
- } else if (primitive.equals(Types.DateType.get())
- && tableSchemaPrimitiveType.equals(Types.TimestampType.withoutZone()))
{
- return Result.DATA_CONVERSION_NEEDED;
- } else if (primitive.typeId() == Type.TypeID.DECIMAL
- && tableSchemaPrimitiveType.typeId() == Type.TypeID.DECIMAL) {
- Types.DecimalType dataType = (Types.DecimalType) primitive;
- Types.DecimalType tableType = (Types.DecimalType)
tableSchemaPrimitiveType;
- return dataType.scale() == tableType.scale() && dataType.precision() <
tableType.precision()
- ? Result.DATA_CONVERSION_NEEDED
- : Result.SCHEMA_UPDATE_NEEDED;
} else {
return Result.SCHEMA_UPDATE_NEEDED;
}
}
+ /**
+ * Whether {@link DataConverter} can convert input data of type {@code
dataType} into the table's
+ * {@code tableType}. Must stay in sync with the conversions {@link
DataConverter#get} performs.
+ */
+ @SuppressWarnings("checkstyle:CyclomaticComplexity")
+ static boolean isDataConversionPossible(
+ Type.PrimitiveType dataType, Type.PrimitiveType tableType) {
+ if (dataType.equals(Types.IntegerType.get()) &&
tableType.equals(Types.LongType.get())) {
+ return true;
+ }
+
+ if (dataType.equals(Types.FloatType.get()) &&
tableType.equals(Types.DoubleType.get())) {
+ return true;
+ }
+
+ if (dataType.equals(Types.DateType.get())
+ && tableType.equals(Types.TimestampType.withoutZone())) {
+ return true;
+ }
Review Comment:
I'm ok fixing/testing the dynsink with nano in a different PR.
Spec promotion rules are a bigger task, as they might have requirements
outside Flink
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
