This is an automated email from the ASF dual-hosted git repository.
voonhous pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new e4718be30790 fix(schema): require a per-field override to promote a
bare long to a timestamp logical type (#19384)
e4718be30790 is described below
commit e4718be307900b5348f26bc113885f224c33896c
Author: Y Ethan Guo <[email protected]>
AuthorDate: Mon Aug 3 21:38:48 2026 -0700
fix(schema): require a per-field override to promote a bare long to a
timestamp logical type (#19384)
Promoting a bare long column to a timestamp logical type during
writer-schema deduction is
now uniformly gated behind the per-field override
hoodie.write.timestamp.logical.type.overrides:
rejected with an actionable error when the field has no override, and
applied when it does.
This holds for all four target types (timestamp-micros, timestamp-millis,
local-timestamp-micros, local-timestamp-millis) and in both reconcile paths
(reconcileSchema
and reconcileTimestampLogicalType).
Previously, long to local-timestamp was already override-gated, but long to
UTC-timestamp was
silently allowed on the default (non-reconcile) write path, because
isGatedTimestampChange did
not treat it as a gated change. A bare long carries no precision signal
(millis vs micros), so
silently attaching a UTC timestamp logical type could mislabel stored
values. This makes the two
cases behave identically. Timestamp precision flips are unchanged.
timestampPrecisionChangeError is made public so tests assert the exact
message without
duplicating its format.
Tests: TestSchemaChangeUtils and TestAvroSchemaEvolutionUtils cover the
gating on both reconcile
paths, with and without an override, for all four targets including nested
fields;
TestHoodieDeltaStreamer.testLongToTimestampPromotionGated exercises the
promotion end to end.
---
.../hudi/common/config/HoodieCommonConfig.java | 7 +-
.../internal/utils/AvroSchemaEvolutionUtils.java | 7 +-
.../schema/internal/utils/SchemaChangeUtils.java | 13 +--
.../utils/TestAvroSchemaEvolutionUtils.java | 100 +++++++++++++++++++++
.../internal/utils/TestSchemaChangeUtils.java | 29 ++++++
.../deltastreamer/TestHoodieDeltaStreamer.java | 98 ++++++++++++++++++++
6 files changed, 244 insertions(+), 10 deletions(-)
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/config/HoodieCommonConfig.java
b/hudi-common/src/main/java/org/apache/hudi/common/config/HoodieCommonConfig.java
index 6724128d3e7f..47e967ad6d08 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/config/HoodieCommonConfig.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/config/HoodieCommonConfig.java
@@ -94,9 +94,10 @@ public class HoodieCommonConfig extends HoodieConfig {
+ "entry is pinned to that logical type: an incoming value of a
different precision is coerced to it, and "
+ "the change from the table's current type is permitted. A
timestamp precision change with no entry for "
+ "the field is rejected with an error, so an unverified
micros/millis flip can never happen silently. "
- + "An entry also attaches a local-timestamp logical type to a column
that 0.x persisted as a bare long "
- + "because its converter did not recognize the type. A UTC/local
zone change is never authorized by "
- + "this config, whatever the entry says, since no rescale can
express it. "
+ + "An entry also attaches a timestamp logical type (UTC or local) to
a column persisted as a bare "
+ + "long, including one that 0.x stored without a logical type
because its converter did not recognize "
+ + "it. A UTC/local zone change is never authorized by this config,
whatever the entry says, since no "
+ + "rescale can express it. "
+ "Derive the value from the stored longs, never from the incoming
schema: for instants after 1990 an "
+ "epoch-millis value is around 1e12 while epoch-micros is around
1e15, so the two ranges do not "
+ "overlap. TimestampLogicalTypeClassifier implements that verdict
for inspection tooling to reuse. "
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/schema/internal/utils/AvroSchemaEvolutionUtils.java
b/hudi-common/src/main/java/org/apache/hudi/common/schema/internal/utils/AvroSchemaEvolutionUtils.java
index 947fc72df797..0310ccfb236e 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/schema/internal/utils/AvroSchemaEvolutionUtils.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/schema/internal/utils/AvroSchemaEvolutionUtils.java
@@ -277,7 +277,12 @@ public class AvroSchemaEvolutionUtils {
col, from, to,
HoodieCommonConfig.TIMESTAMP_LOGICAL_TYPE_OVERRIDES.key()));
}
- private static SchemaCompatibilityException
timestampPrecisionChangeError(String col, Type from, Type to) {
+ /**
+ * Builds the actionable error for a gated timestamp logical-type change
with no per-field override
+ * in {@code hoodie.write.timestamp.logical.type.overrides}. Public so tests
can assert the exact
+ * message without duplicating its format.
+ */
+ public static SchemaCompatibilityException
timestampPrecisionChangeError(String col, Type from, Type to) {
return new SchemaCompatibilityException(String.format(
"Refusing to change the timestamp logical type of column '%s' from
'%s' to '%s' without an explicit "
+ "verdict. This precision change is not applied automatically
because the correct target depends "
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/schema/internal/utils/SchemaChangeUtils.java
b/hudi-common/src/main/java/org/apache/hudi/common/schema/internal/utils/SchemaChangeUtils.java
index 567c1397f431..f7d8ba142dde 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/schema/internal/utils/SchemaChangeUtils.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/schema/internal/utils/SchemaChangeUtils.java
@@ -96,8 +96,8 @@ public class SchemaChangeUtils {
/**
* Whether a column type change is a timestamp precision change that must be
authorized by an
* explicit per-field override (see {@code
hoodie.write.timestamp.logical.type.overrides}). This
- * covers timestamp-micros/millis flips, local-timestamp-micros/millis
flips, and the forward-fix
- * from a bare {@code long} to a local-timestamp logical type that 0.x
dropped.
+ * covers timestamp-micros/millis flips, local-timestamp-micros/millis
flips, and promoting a bare
+ * {@code long} to a timestamp logical type (UTC or local).
*/
public static boolean isGatedTimestampChange(Type src, Type dst) {
if (src.equals(dst)) {
@@ -109,7 +109,7 @@ public class SchemaChangeUtils {
if (isLocalTimestamp(src) && isLocalTimestamp(dst)) {
return true;
}
- return src.typeId() == Type.TypeID.LONG && isLocalTimestamp(dst);
+ return src.typeId() == Type.TypeID.LONG && (isUtcTimestamp(dst) ||
isLocalTimestamp(dst));
}
/**
@@ -170,9 +170,10 @@ public class SchemaChangeUtils {
|| dst == Types.DoubleType.get() || dst == Types.StringType.get()
|| dst.typeId() == Type.TypeID.DECIMAL || dst.typeId() ==
Type.TypeID.DECIMAL_FIXED;
case LONG:
if (allowTimestampPrecisionEvolution
- && (dst.typeId() == Type.TypeID.LOCAL_TIMESTAMP_MILLIS ||
dst.typeId() == Type.TypeID.LOCAL_TIMESTAMP_MICROS)) {
- // Forward-fix path: 0.x stored local-timestamp columns as bare long
because its converter
- // did not recognize the logical type. Allow attaching the logical
type when the gate is open.
+ && (dst.typeId() == Type.TypeID.TIMESTAMP || dst.typeId() ==
Type.TypeID.TIMESTAMP_MILLIS
+ || dst.typeId() == Type.TypeID.LOCAL_TIMESTAMP_MILLIS ||
dst.typeId() == Type.TypeID.LOCAL_TIMESTAMP_MICROS)) {
+ // A bare long carries no precision signal, so promoting it to a
timestamp logical type is
+ // authorized only by an explicit per-field override.
return true;
}
return dst == Types.FloatType.get() || dst == Types.DoubleType.get()
|| dst == Types.StringType.get() || dst.typeId() == Type.TypeID.DECIMAL ||
dst.typeId() == Type.TypeID.DECIMAL_FIXED;
diff --git
a/hudi-common/src/test/java/org/apache/hudi/common/schema/internal/utils/TestAvroSchemaEvolutionUtils.java
b/hudi-common/src/test/java/org/apache/hudi/common/schema/internal/utils/TestAvroSchemaEvolutionUtils.java
index 83f0e47af648..92dc75303780 100644
---
a/hudi-common/src/test/java/org/apache/hudi/common/schema/internal/utils/TestAvroSchemaEvolutionUtils.java
+++
b/hudi-common/src/test/java/org/apache/hudi/common/schema/internal/utils/TestAvroSchemaEvolutionUtils.java
@@ -51,6 +51,7 @@ import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;
+import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -840,4 +841,103 @@ public class TestAvroSchemaEvolutionUtils {
SchemaChangeUtils.parseTimestampLogicalTypeOverrides("ts:local-timestamp-millis")).toAvroSchema();
Assertions.assertEquals("local-timestamp-millis",
stillWorks.getField("ts").schema().getLogicalType().getName());
}
+
+ @Test
+ void testLongToUtcTimestampGatedInBothReconcilePaths() {
+ // Bare long to a UTC timestamp is override-gated exactly like the
local-timestamp case: rejected
+ // without a per-field override and applied with one, in both reconcile
paths. The non-reconcile
+ // guard previously skipped this and let it through silently on the
default write path.
+ HoodieSchema tableBareLong =
HoodieSchema.fromAvroSchema(tripAvro(Schema.create(Schema.Type.LONG)));
+ HoodieSchema incomingMicros =
HoodieSchema.fromAvroSchema(tripAvro(LogicalTypes.timestampMicros().addToSchema(Schema.create(Schema.Type.LONG))));
+
+ // No override: rejected in both paths with the exact actionable error.
+ Map<String, Type> noOverride =
SchemaChangeUtils.parseTimestampLogicalTypeOverrides("");
+ String expectedError =
AvroSchemaEvolutionUtils.timestampPrecisionChangeError(
+ "ts", Types.LongType.get(), Types.TimestampType.get()).getMessage();
+ SchemaCompatibilityException reconcileError =
assertThrows(SchemaCompatibilityException.class,
+ () -> AvroSchemaEvolutionUtils.reconcileSchema(incomingMicros,
tableBareLong, false, noOverride));
+ assertEquals(expectedError, reconcileError.getMessage());
+ SchemaCompatibilityException guardError =
assertThrows(SchemaCompatibilityException.class,
+ () ->
AvroSchemaEvolutionUtils.reconcileTimestampLogicalType(incomingMicros,
tableBareLong, noOverride));
+ assertEquals(expectedError, guardError.getMessage());
+
+ // With the override: the promotion is applied in both paths.
+ Schema viaReconcile =
AvroSchemaEvolutionUtils.reconcileSchema(incomingMicros, tableBareLong, false,
+
SchemaChangeUtils.parseTimestampLogicalTypeOverrides("ts:timestamp-micros")).toAvroSchema();
+ assertEquals("timestamp-micros",
viaReconcile.getField("ts").schema().getLogicalType().getName());
+ Schema viaGuard =
AvroSchemaEvolutionUtils.reconcileTimestampLogicalType(incomingMicros,
tableBareLong,
+
SchemaChangeUtils.parseTimestampLogicalTypeOverrides("ts:timestamp-micros")).toAvroSchema();
+ assertEquals("timestamp-micros",
viaGuard.getField("ts").schema().getLogicalType().getName());
+ }
+
+ @Test
+ void testLongToLocalTimestampGatedInBothReconcilePaths() {
+ // Bare long to local timestamp is override-gated (not forbidden):
rejected without an override
+ // and applied with one, and the non-reconcile guard must agree with
reconcileSchema.
+ HoodieSchema tableBareLong =
HoodieSchema.fromAvroSchema(tripAvro(Schema.create(Schema.Type.LONG)));
+ HoodieSchema incomingLocalMicros =
HoodieSchema.fromAvroSchema(tripAvro(LogicalTypes.localTimestampMicros().addToSchema(Schema.create(Schema.Type.LONG))));
+
+ // No override: rejected in both paths with the exact actionable error.
+ Map<String, Type> noOverride =
SchemaChangeUtils.parseTimestampLogicalTypeOverrides("");
+ String expectedError =
AvroSchemaEvolutionUtils.timestampPrecisionChangeError(
+ "ts", Types.LongType.get(),
Types.LocalTimestampMicrosType.get()).getMessage();
+ SchemaCompatibilityException reconcileError =
assertThrows(SchemaCompatibilityException.class,
+ () -> AvroSchemaEvolutionUtils.reconcileSchema(incomingLocalMicros,
tableBareLong, false, noOverride));
+ assertEquals(expectedError, reconcileError.getMessage());
+ SchemaCompatibilityException guardError =
assertThrows(SchemaCompatibilityException.class,
+ () ->
AvroSchemaEvolutionUtils.reconcileTimestampLogicalType(incomingLocalMicros,
tableBareLong, noOverride));
+ assertEquals(expectedError, guardError.getMessage());
+ Schema repaired =
AvroSchemaEvolutionUtils.reconcileTimestampLogicalType(incomingLocalMicros,
tableBareLong,
+
SchemaChangeUtils.parseTimestampLogicalTypeOverrides("ts:local-timestamp-micros")).toAvroSchema();
+ assertEquals("local-timestamp-micros",
repaired.getField("ts").schema().getLogicalType().getName());
+ }
+
+ @Test
+ void testNestedLongToTimestampGated() {
+ // The gate resolves fully-qualified column names, so it applies to nested
fields too. A nested
+ // long -> timestamp (UTC or local) is override-gated via the dotted-key
override.
+ for (String token : new String[] {"timestamp-micros",
"local-timestamp-millis"}) {
+ HoodieSchema tableNested =
HoodieSchema.fromAvroSchema(nestedTrip(Schema.create(Schema.Type.LONG)));
+ HoodieSchema incoming =
HoodieSchema.fromAvroSchema(nestedTrip(logicalLong(token)));
+ // No override: rejected in both paths with the exact actionable error.
+ Map<String, Type> noOverride =
SchemaChangeUtils.parseTimestampLogicalTypeOverrides("");
+ String expectedError =
AvroSchemaEvolutionUtils.timestampPrecisionChangeError("payload.event_ts",
Types.LongType.get(),
+ SchemaChangeUtils.parseTimestampLogicalTypeOverrides("field:" +
token).get("field")).getMessage();
+ SchemaCompatibilityException reconcileError =
assertThrows(SchemaCompatibilityException.class,
+ () -> AvroSchemaEvolutionUtils.reconcileSchema(incoming,
tableNested, false, noOverride));
+ assertEquals(expectedError, reconcileError.getMessage());
+ SchemaCompatibilityException guardError =
assertThrows(SchemaCompatibilityException.class,
+ () ->
AvroSchemaEvolutionUtils.reconcileTimestampLogicalType(incoming, tableNested,
noOverride));
+ assertEquals(expectedError, guardError.getMessage());
+ // The dotted-key override authorizes the nested promotion.
+ Schema repairedNested =
AvroSchemaEvolutionUtils.reconcileTimestampLogicalType(incoming, tableNested,
+
SchemaChangeUtils.parseTimestampLogicalTypeOverrides("payload.event_ts:" +
token)).toAvroSchema();
+ assertEquals(token,
+
repairedNested.getField("payload").schema().getField("event_ts").schema().getLogicalType().getName());
+ }
+ }
+
+ private static Schema nestedTrip(Schema eventTsType) {
+ Schema payload = Schema.createRecord("payloadrec", null, null, false,
Arrays.asList(
+ new Schema.Field("event_ts", eventTsType, null, null)));
+ return Schema.createRecord("trip", null, null, false, Arrays.asList(
+ new Schema.Field("id", Schema.create(Schema.Type.STRING), null, null),
+ new Schema.Field("payload", payload, null, null)));
+ }
+
+ private static Schema logicalLong(String token) {
+ Schema longSchema = Schema.create(Schema.Type.LONG);
+ switch (token) {
+ case "timestamp-micros":
+ return LogicalTypes.timestampMicros().addToSchema(longSchema);
+ case "timestamp-millis":
+ return LogicalTypes.timestampMillis().addToSchema(longSchema);
+ case "local-timestamp-micros":
+ return LogicalTypes.localTimestampMicros().addToSchema(longSchema);
+ case "local-timestamp-millis":
+ return LogicalTypes.localTimestampMillis().addToSchema(longSchema);
+ default:
+ throw new IllegalArgumentException(token);
+ }
+ }
}
diff --git
a/hudi-common/src/test/java/org/apache/hudi/common/schema/internal/utils/TestSchemaChangeUtils.java
b/hudi-common/src/test/java/org/apache/hudi/common/schema/internal/utils/TestSchemaChangeUtils.java
index 9e0579214bea..522e2559a967 100644
---
a/hudi-common/src/test/java/org/apache/hudi/common/schema/internal/utils/TestSchemaChangeUtils.java
+++
b/hudi-common/src/test/java/org/apache/hudi/common/schema/internal/utils/TestSchemaChangeUtils.java
@@ -26,6 +26,7 @@ import org.junit.jupiter.api.Test;
import java.util.Map;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -120,4 +121,32 @@ public class TestSchemaChangeUtils {
assertThrows(UnsupportedOperationException.class,
() -> overrides.put("other", Types.TimestampMillisType.get()));
}
+
+ @Test
+ void gatedTimestampChangeCoversFlipsAndLongPromotions() {
+ // Precision flips (either direction) are gated.
+
assertTrue(SchemaChangeUtils.isGatedTimestampChange(Types.TimestampType.get(),
Types.TimestampMillisType.get()));
+
assertTrue(SchemaChangeUtils.isGatedTimestampChange(Types.LocalTimestampMicrosType.get(),
Types.LocalTimestampMillisType.get()));
+ // Promoting a bare long to any timestamp logical type (UTC or local) is
gated.
+ assertTrue(SchemaChangeUtils.isGatedTimestampChange(Types.LongType.get(),
Types.TimestampType.get()));
+ assertTrue(SchemaChangeUtils.isGatedTimestampChange(Types.LongType.get(),
Types.TimestampMillisType.get()));
+ assertTrue(SchemaChangeUtils.isGatedTimestampChange(Types.LongType.get(),
Types.LocalTimestampMillisType.get()));
+ assertTrue(SchemaChangeUtils.isGatedTimestampChange(Types.LongType.get(),
Types.LocalTimestampMicrosType.get()));
+ // Unrelated promotions and identical types are not gated.
+ assertFalse(SchemaChangeUtils.isGatedTimestampChange(Types.LongType.get(),
Types.StringType.get()));
+ assertFalse(SchemaChangeUtils.isGatedTimestampChange(Types.IntType.get(),
Types.TimestampType.get()));
+
assertFalse(SchemaChangeUtils.isGatedTimestampChange(Types.TimestampType.get(),
Types.TimestampType.get()));
+ }
+
+ @Test
+ void typeUpdateAllowGatesLongToTimestampBehindTheOverride() {
+ // Promoting a bare long to any timestamp logical type is allowed only
when the gate is open.
+ for (Type ts : new Type[] {Types.TimestampType.get(),
Types.TimestampMillisType.get(),
+ Types.LocalTimestampMillisType.get(),
Types.LocalTimestampMicrosType.get()}) {
+ assertTrue(SchemaChangeUtils.isTypeUpdateAllow(Types.LongType.get(), ts,
true));
+ assertFalse(SchemaChangeUtils.isTypeUpdateAllow(Types.LongType.get(),
ts, false));
+ }
+ // Existing long widening is unaffected by the gate.
+ assertTrue(SchemaChangeUtils.isTypeUpdateAllow(Types.LongType.get(),
Types.DoubleType.get(), false));
+ }
}
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieDeltaStreamer.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieDeltaStreamer.java
index 86165740dcbc..16aeb35971e5 100644
---
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieDeltaStreamer.java
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieDeltaStreamer.java
@@ -54,6 +54,10 @@ import org.apache.hudi.common.schema.HoodieSchema;
import org.apache.hudi.common.schema.HoodieSchema.TimePrecision;
import org.apache.hudi.common.schema.HoodieSchemaField;
import org.apache.hudi.common.schema.HoodieSchemaType;
+import org.apache.hudi.common.schema.internal.Type;
+import org.apache.hudi.common.schema.internal.Types;
+import org.apache.hudi.common.schema.internal.utils.AvroSchemaEvolutionUtils;
+import org.apache.hudi.common.schema.internal.utils.SchemaChangeUtils;
import org.apache.hudi.common.table.HoodieTableConfig;
import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.table.HoodieTableVersion;
@@ -132,6 +136,8 @@ import org.apache.hudi.utilities.transform.Transformer;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
+import org.apache.avro.LogicalType;
+import org.apache.avro.LogicalTypes;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.generic.IndexedRecord;
@@ -201,6 +207,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assertions.fail;
@@ -760,6 +767,97 @@ public class TestHoodieDeltaStreamer extends
HoodieDeltaStreamerTestBase {
assertEquals(0,
sqlContext.read().options(hudiOpts).format("org.apache.hudi").load(tableBasePath).filter("current_ts
< '1980-01-01'").count());
}
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void testLongToTimestampPromotionGated(boolean setNullForMissingColumns)
throws Exception {
+ // Promoting a plain long column to a timestamp logical type is
override-gated: rejected without a
+ // per-field override for every target type, and applied with one. A bare
long carries no precision
+ // signal, so the override is the explicit verdict that authorizes the
promotion. One bare-long seed
+ // is reused: the rejection cases all throw (table stays bare long), and
the accepted case runs last.
+ String tableBasePath = basePath + "/testLongToTs" +
setNullForMissingColumns;
+ defaultSchemaProviderClassName = FilebasedSchemaProvider.class.getName();
+
+ // Sync 0: seed the table with `seconds_since_epoch` stored as a bare long.
+ HoodieDeltaStreamer.Config seed = TestHelpers.makeConfig(tableBasePath,
WriteOperationType.INSERT,
+ Collections.singletonList(TestIdentityTransformer.class.getName()),
PROPS_FILENAME_TEST_SOURCE,
+ false, true, false, null, HoodieTableType.COPY_ON_WRITE.name());
+ seed.configs.add("hoodie.streamer.schemaprovider.source.schema.file=" +
basePath + "/source-timestamp-millis.avsc");
+ seed.configs.add("hoodie.streamer.schemaprovider.target.schema.file=" +
basePath + "/source-timestamp-millis.avsc");
+ seed.configs.add("hoodie.datasource.write.row.writer.enable=false");
+ seed.configs.add(HoodieCommonConfig.SET_NULL_FOR_MISSING_COLUMNS.key() +
"=" + setNullForMissingColumns);
+ new HoodieDeltaStreamer(seed, jsc).sync();
+
+ Schema tableSchema = new
TableSchemaResolver(HoodieTestUtils.createMetaClient(storage, tableBasePath))
+ .getTableSchema(false).toAvroSchema();
+
assertNull(tableSchema.getField("seconds_since_epoch").schema().getLogicalType(),
+ "seconds_since_epoch must be seeded as a bare long in the table");
+ Schema baseSchema = new Schema.Parser().parse(fs.open(new Path(basePath +
"/source-timestamp-millis.avsc")));
+
+ // Every target type is rejected without a per-field override.
+ for (LogicalType targetType : new LogicalType[]
{LogicalTypes.timestampMillis(), LogicalTypes.timestampMicros(),
+ LogicalTypes.localTimestampMillis(),
LogicalTypes.localTimestampMicros()}) {
+ String schemaFile = writePromotedSchema(baseSchema, targetType,
setNullForMissingColumns);
+ HoodieDeltaStreamer.Config reject = promoteConfig(tableBasePath,
schemaFile, setNullForMissingColumns, null);
+ HoodieDeltaStreamer streamer = new HoodieDeltaStreamer(reject, jsc);
+ // sync() wraps the guard's SchemaCompatibilityException in a
HoodieIngestionException, so walk
+ // the cause chain to assert on the underlying exception.
+ Throwable thrown = assertThrows(Exception.class, streamer::sync,
+ "long -> " + targetType.getName() + " must be rejected without an
override");
+ Throwable cause = thrown;
+ while (cause != null && !(cause instanceof
SchemaCompatibilityException)) {
+ cause = cause.getCause();
+ }
+ assertTrue(cause instanceof SchemaCompatibilityException,
+ "Expected a SchemaCompatibilityException in the cause chain, got: "
+ thrown);
+ Type toType =
SchemaChangeUtils.parseTimestampLogicalTypeOverrides("field:" +
targetType.getName()).get("field");
+ assertEquals(AvroSchemaEvolutionUtils.timestampPrecisionChangeError(
+ "seconds_since_epoch", Types.LongType.get(), toType).getMessage(),
cause.getMessage());
+ }
+
+ // With an override the promotion is authorized (local promotions are
covered end-to-end by
+ // testCOWLogicalRepair / testMORLogicalRepair); verify a UTC promotion
succeeds and lands on the
+ // table schema.
+ String utcSchemaFile = writePromotedSchema(baseSchema,
LogicalTypes.timestampMicros(), setNullForMissingColumns);
+ HoodieDeltaStreamer.Config accept = promoteConfig(tableBasePath,
utcSchemaFile, setNullForMissingColumns,
+ "seconds_since_epoch:timestamp-micros");
+ new HoodieDeltaStreamer(accept, jsc).sync();
+ Schema evolved = new
TableSchemaResolver(HoodieTestUtils.createMetaClient(storage, tableBasePath))
+ .getTableSchema(false).toAvroSchema();
+ assertEquals("timestamp-micros",
evolved.getField("seconds_since_epoch").schema().getLogicalType().getName());
+ }
+
+ private String writePromotedSchema(Schema baseSchema, LogicalType
targetType, boolean setNull) throws IOException {
+ Schema incoming = replaceFieldType(baseSchema, "seconds_since_epoch",
+ targetType.addToSchema(Schema.create(Schema.Type.LONG)));
+ String schemaFile = basePath + "/promote-" + targetType.getName() + "-nul"
+ setNull + ".avsc";
+ UtilitiesTestBase.Helpers.saveStringsToDFS(new String[]
{incoming.toString()}, storage, schemaFile);
+ return schemaFile;
+ }
+
+ private HoodieDeltaStreamer.Config promoteConfig(String tableBasePath,
String schemaFile,
+ boolean
setNullForMissingColumns, String override) {
+ HoodieDeltaStreamer.Config cfg = TestHelpers.makeConfig(tableBasePath,
WriteOperationType.UPSERT,
+ Collections.singletonList(TestIdentityTransformer.class.getName()),
PROPS_FILENAME_TEST_SOURCE,
+ false, true, false, null, HoodieTableType.COPY_ON_WRITE.name());
+ cfg.configs.add("hoodie.streamer.schemaprovider.source.schema.file=" +
schemaFile);
+ cfg.configs.add("hoodie.streamer.schemaprovider.target.schema.file=" +
schemaFile);
+ cfg.configs.add("hoodie.datasource.write.row.writer.enable=false");
+ cfg.configs.add(HoodieCommonConfig.SET_NULL_FOR_MISSING_COLUMNS.key() +
"=" + setNullForMissingColumns);
+ if (override != null) {
+
cfg.configs.add(HoodieCommonConfig.TIMESTAMP_LOGICAL_TYPE_OVERRIDES.key() + "="
+ override);
+ }
+ return cfg;
+ }
+
+ private static Schema replaceFieldType(Schema recordSchema, String
fieldName, Schema newFieldType) {
+ List<Schema.Field> fields = new ArrayList<>();
+ for (Schema.Field field : recordSchema.getFields()) {
+ Schema fieldSchema = field.name().equals(fieldName) ? newFieldType :
field.schema();
+ fields.add(new Schema.Field(field.name(), fieldSchema, field.doc(),
field.defaultVal()));
+ }
+ return Schema.createRecord(recordSchema.getName(), recordSchema.getDoc(),
recordSchema.getNamespace(), false, fields);
+ }
+
@Test
public void testLogicalTypes() throws Exception {
try {
