Core: Support local-timestamp-* in AvroSchemaUtil

[kinolaev]

Related:

• issue Support both adjust-to-utc and local-timestamp-micros in Avro Data Type Mappings #11903
• issue Spark, Avro and Iceberg Timestamp, Timestamp NTZ definition need more clarifications #12751
• PR Avro: Encode non-zone timestamps with local-timestamp logical types #16577
• discussion thread https://lists.apache.org/thread/jncwontk4xkmt7n5ml0pbgk4x54cwzgo
• PR Core: Support reading Avro local-timestamp-* logical types #15437
• PR Core, Kafka Connect: Support additional Avro temporal logical types #16681

Unlike similar PRs, this one has a very narrow scope and does not introduce any breaking changes.

With legacyTimestampMapping flag set to false, the AvroSchemaUtil.toIcebergSchema method produces an Iceberg schema that is compatible with Flink's AvroSchemaConverter output (convertToTypeInfo, convertToDataType).

[kinolaev]

Also added legacyTimestampMapping to TypeToSchema to support local-timestamp-* types during Iceberg-to-Avro conversions.

[AnatolyPopov]

Could you clarify which related PRs are considered breaking here? PR #15437 did not change Iceberg’s Avro write encoding or any public API. It only recognized local-timestamp-* on input and updated the reader paths that previously rejected those logical types.

I’m also concerned that legacyTimestampMapping=false does more than enable standard local timestamps. It causes every timestamp-* to map to a zoned Iceberg timestamp, even when the schema contains adjust-to-utc=false. Therefore, the new mode cannot safely read schemas containing both legacy Iceberg timestamps and Avro local-timestamp-*.

For backward compatibility, Avro to Iceberg conversion should continue honoring adjust-to-utc while recognizing local-timestamp-* as timestamps without zone. These behaviors are not mutually exclusive and were both supported by #15437.

Also, no production caller currently passes false, and the actual Iceberg Avro readers/writers still reject local-timestamp-*. Is this PR intentionally limited to an external schema-conversion API rather than end-to-end read/write support?

[kinolaev]

Could you clarify which related PRs are considered breaking here?

All three PRs unconditionally add support for local-timestamp-* types to SchemaToType. This means the output of methods like AvroSchemaUtil.toIcebergSchema will change after an upgrade (long columns will become timestamps), which could break downstream pipelines.

Avro to Iceberg conversion should continue honoring adjust-to-utc

To avoid this breaking change, I actually prefer having two separate modes:

1. Iceberg-specific: Ignores local-timestamp-* and honors the adjust-to-utc property.
2. Avro-native: Honors logical types and ignores the adjust-to-utc property.

By the way, local-timestamp-* types are already supported in Flink AvroSchemaConverter and AvroToRowDataConverter when legacyTimestampMapping is set to false. I decided to adapt the same approach for AvroSchemaUtil.

Is this PR intentionally limited to an external schema-conversion API rather than end-to-end read/write support?

Yes, it is intentionally limited. I recently started a project to ingest data from Kafka using the Flink Dynamic Iceberg Sink. Kafka Connect serializes the data into Avro format, and Iceberg Flink converts the Avro GenericRecord to an Iceberg Schema and Flink RowData. While this seemed straightforward at first, I ran into an issue with timestamp handling. Since Flink already has an Avro-native mode, I'm only looking to add support for Avro-native mode in AvroSchemaUtil for now.

[kinolaev]

Avro to Iceberg conversion should continue honoring adjust-to-utc while recognizing local-timestamp-* as timestamps without zone. These behaviors are not mutually exclusive

I think they are exclusive in practice. A writer writes either the adjust-to-utc=false property or the local-timestamp-* logical type. A reader must know the writer's mode in advance to correctly interpret a timestamp-* schema without adjust-to-utc property, because AvroSchemaUtil.isTimestamptz returns false in these cases.

[kinolaev]

For backward compatibility, Avro to Iceberg conversion should continue honoring adjust-to-utc

I made Avro-native mode backward compatible with Iceberg-specific mode in fa90be8. With legacyTimestampMapping=false the adjust-to-utc property is honored but defaults to true if not set.

Does it address your concern, @AnatolyPopov ?

[AnatolyPopov]

Oh yeah, now I see what you meant by the breaking change. But do we know any Iceberg engine path or downstream code that relies on the converted type being LongType? I saw this more as missing support for the logical type than an established contract and what I've seen in some places before is that logical type in this case was not recognized and underlying physical long type was used as a fallback.

The other concern is addressed now, thanks! But it still seems to be missing tests for the new behavior. Could you add tests for legacyTimestampMapping=false with adjust-to-utc=false and with the property missing? A mixed-schema test would be useful too.

[kinolaev]

But do we know any Iceberg engine path or downstream code that relies on the converted type being LongType? I saw this more as missing support for the logical type than an established contract

While I agree that this is more of a missing feature, I would prefer not to make any assumptions about public API users. In a sense, even a missing feature in a public API is an established contract)

I've updated the PR. TypeToSchema (Iceberg -> Avro) no longer sets adjust-to-utc when legacyTimestampMapping=false. I made this decision because a user must switch SchemaToType to legacyTimestampMapping=false first anyway to be able to convert local-timestamp-*. Once SchemaToType is switched, I see no reason to add adjust-to-utc=true after switching TypeToSchema.

This change is reflected in the updated test. Iceberg-specific and Avro-native timestamp conversions are now covered.

Thanks for the review, @AnatolyPopov !

[AnatolyPopov]

Looks good to me know, thank for addressing the comments and hopefully this time the support local timestamps will come though. There were many attempts already.

[AnatolyPopov]

@RussellSpitzer I've seen you had some concerns regarding similar things before. Would you mind to take a look?

[RussellSpitzer]

These branches will fall through and return null if legacyTimeMapping is true. I think the assumption was that the reader is assuming legacy status of the writer, but i'm not sure you can do that. For example if my writer is set to "non-legacy" but my reader is set to "legacy" then it will break.

Consider the following test which would currently fail

@Test
public void testLocalTimestampWithLegacyMapping() {
    Schema localTsMicros =
        LogicalTypes.localTimestampMicros().addToSchema(Schema.create(Schema.Type.LONG));
    Schema localTsNanos =
        LogicalTypes.localTimestampNanos().addToSchema(Schema.create(Schema.Type.LONG));

    // local-timestamp-* types are semantically unambiguous — always no timezone.
    // legacyTimestampMapping should have no effect on them.
    assertThat(AvroSchemaUtil.convert(localTsMicros))
        .isEqualTo(Types.TimestampType.withoutZone());
    assertThat(AvroSchemaUtil.convert(localTsNanos))
        .isEqualTo(Types.TimestampNanoType.withoutZone());
}

I Think this is fixed by just dropping the condition here (if !legacy)

[kinolaev]

Dropping the condition introduces a breaking change: if someone uses AvroSchemaUtil.toIcebergSchema to write externally produced Avro data with local-timestamp-*, their pipeline will fail after the upgrade because the long columns will become timestamps. We discussed this above #17196 (comment). Does it look like a breaking change to you, @RussellSpitzer ? Do you prefer to support local-timestamp-* unconditionally?

I don't know any real consumer that relies on ignoring local-timestamp-* and I'm ready to drop the condition. Just wanted to highlight this first.

[RussellSpitzer]

I'm not sure I understand. How would we produce local-timestamp-* after they upgrade? It would only be produced by writers with the legacy flag off which would not be the default?

This is guarding against users who were manually creating local-timestamp in their Avro schemas (which I think has to come from outside our library) but are relying on this function to return a "long" in that use-case.

There is an issue with "timestamp-micros" (not local) but that is handled correctly by passing through the legacy mode

[kinolaev]

This is guarding against users who were manually creating local-timestamp in their Avro schemas (which I think has to come from outside our library) but are relying on this function to return a "long" in that use-case.

Exactly. Do we care about this rather unlikely use-case?

[RussellSpitzer]

This is "core" and not "api" so I'd rather we not support what we know is an incorrect behavior imho. We can always send out a dev list thread noting the change but honestly it feel's like supporting a bug to me

[AnatolyPopov]

This is "core" and not "api" so I'd rather we not support what we know is an incorrect behavior imho. We can always send out a dev list thread noting the change but honestly it feel's like supporting a bug to me

I would rather agree to this.
I faced exactly this case when I started working on a previous outdated PR for this issue and concluded that it does not make sense to fully rely on the return type being long, except as a temporary fallback before this is implemented.

[kinolaev]

I added the condition to avoid discussions about breaking changes, but now I look like the most conservative person in the room! 😄 No problem, I'll drop the condition.

[kinolaev]

The condition is dropped (0661934)

[RussellSpitzer]

Started a thread here https://lists.apache.org/thread/qrr6hwzy70slxz24s3gr5dz68mxys9ls

[kinolaev]

Thanks @RussellSpitzer !

There was a request on the PR to preserve this behavior in the API via a parameter

To be precise, I'm not requesting to preserve this behavior. As I mentioned above, I added the condition for local-timestamp-* only to avoid discussions about breaking changes. Personally, I am all for unconditional local-timestamp-* support!

[RussellSpitzer]

I think this approach is trying to solve essetnially two problems at the same time but I think they should be fixed in different ways.

On Read, we shouldn't have any optional behavior, if you see the new types you see the new types. Not an issue. So there we just need to add support for the new types.

Reads

1. See local timestamp - use withoutZone
2. timestamp and adjust-to-utc
3. timestamp and no prop - Check table Config

On the write side
Instead of a parameter we plumb through all these methods, we should just have a table property similar to AVRO_COMPRESSION and a private method which takes it's resolution. Something like

// In TableProperties:
String AVRO_TIMESTAMP_ENCODING = "write.avro.timestamp-encoding";
String AVRO_TIMESTAMP_ENCODING_DEFAULT = "legacy";  // or "local-timestamp"

// In Avro.WriteBuilder.build():
boolean legacy = !"local-timestamp".equals(config.get(AVRO_TIMESTAMP_ENCODING));
schema = AvroSchemaUtil.convert(icebergSchema, name, legacy);  // internal call, not public

[RussellSpitzer]

Please don't take my code above as gospel. Thinking about it for a half second more I think we should probably keep it a boolean and not a string...

// In TableProperties:
String AVRO_TIMESTAMP_ENCODING_LEGACY_MODE = "write.avro.timestamp-encoding";
Boolean AVRO_TIMESTAMP_ENCODING_LEGACY_MODE_DEFAULT = true;  // or false (use local-timestamps)

[kinolaev]

boolean legacy = !"local-timestamp".equals(config.get(AVRO_TIMESTAMP_ENCODING));
schema = AvroSchemaUtil.convert(icebergSchema, name, legacy); // internal call, not public

I'd like to use AvroSchemaUtil.convert in my own project to convert external Avro data and write it into Iceberg, like in this example https://iceberg.apache.org/docs/latest/flink-writes/#write-with-avro-genericrecord. Not to add Avro data files to an Iceberg table. So I'd like to have a public method that supports local-timestamp-* and converts timestamp-* without adjust-to-utc property to timestamptz(_ns). So for me a table property is not an option.

@RussellSpitzer , is adding a public method still a possibility we can consider?

[RussellSpitzer]

Let's start a general dev list thread. My gut instinct here is no, we don't want to keep annotating this method with legacy flags because we don't actually want to support that behavior for outside consumers. If you use this method and we see an Iceberg type we know, we should get the right Iceberg type.

For precedent though you can check out
#12455

Where we similarly added recognition for new types (timestamp 9) which previously would have fell through and become Long.

[kinolaev]

Even after dropping the condition for local-timestamp-*, setting legacyTimestampMapping to false still leads to the following differences compared to legacyTimestampMapping=true:

1. Avro timestamp-* without the adjust-to-utc property is converted to Iceberg timestamptz(_ns) (instead of timestamp(_ns))
2. Iceberg timestamptz(_ns) is converted to Avro timestamp-* without the adjust-to-utc property (instead of timestamp-* with adjust-to-utc=true)
3. Iceberg timestamp(_ns) is converted to Avro local-timestamp-* (instead of timestamp-* with adjust-to-utc=false)

So, the question is not, "Do we want Avro local-timestamp-* to Iceberg timestamp(_ns) conversion support to be flagged with legacyTimestampMapping?". Rather, the question is, "Do we want to provide public methods that follow Avro semantics during round-trip conversions"?

Are we on the same page, @RussellSpitzer ?

upd: I saw your message below. Let's continue this discussion in the main thread.

[RussellSpitzer]

There are a few consumers that need to be updated so they don't break on the new types.

DataReader.java:163-164
PlannedDataReader.java:171-172
DataWriter.java:140-141

[RussellSpitzer]

There are some other downstream reader issues. DataReader.isTimestampZ hardcodes a "true" in the legacy mode parameter. The writer not in legacy mode would always be read incorrectly.

[RussellSpitzer]

I don't think we need to supress the warning here. We can just extract the timestamp logic out into it's own helper if required after removing the !legacy checks.

[kinolaev]

Added helpers and removed warning suppression (0661934)

[RussellSpitzer]

This looks like 3 independent tests here to me. I'd split these up

[kinolaev]

There are several other tests in TestSchemaConversions that combine avro->iceberg and iceberg->avro conversions in a single test, like testStructAndPrimitiveTypes. Please confirm if you want me to separate them.

[RussellSpitzer]

I didn't review the other tests :) but let's take a deeper look

Well if we really look into this we have two tests.

1. Roundtrip on the schema
2. Test of oneway legacy schema conversion to Iceberg

The "roundtrip test" here is redundant with the "testPrimitiveTypes" round trip testing above. So we can drop that entirely and make sure it's covered above. Then this keeps the one-way test and trims it down to just the legacyAvroSchema => icebergSchema

[kinolaev]

I don't agree that the round trip is redundant because Avro timestamps in testPrimitiveTypes and testTimestampTypesWithLegacyMappingDisabled have different schemas.

I've added nano timestamps to testPrimitiveTypes. I've also added cases that are not covered by the round trips for both - legacy and Avro native - modes to separate tests: testAvroToIcebergTimestampTypes and testAvroToIcebergTimestampTypesWithLegacyMappingDisabled. I hope, all conversions are covered now.

5de86af

[RussellSpitzer]

A bit above here in testPrimitiveTypes we are missing entries for TimestampNanos.with and withoutZone. (Also check out addAdjustToUtc) I'm just noting this becasue we now only have tests for these with legacy mode false.

[kinolaev]

Added (5de86af)

[RussellSpitzer]

Stylistically I tink we generally don't want final on local vars

[kinolaev]

Fixed (0661934)

[RussellSpitzer]

I'd rather we use callsite selection of schema, but if we keep this the assignments should be

this.private_field = new_private_field_value

for Iceberg style

[kinolaev]

Moved from the constructor (0661934)

[RussellSpitzer]

nit: And i'm sorry this isn't in checkstyle, we fight about this alot. But there should be a linebreak after the } brace.

[kinolaev]

A linebreak added (0661934)

[kinolaev]

I've dropped the condition for local-timestamp-* types, updated the tests, and addressed style-related issues.

All checks have passed. It looks like enabling local-timestamp-* type support in AvroSchemaUtil doesn't break DataReader, PlannedDataReader and DataWriter. Can we keep this PR scoped to the AvroSchemaUtil class?

The main question remains: do we want to introduce public methods on AvroSchemaUtil that:

1. convert Avro timestamp-* to Iceberg timestamptz(_ns) when the adjust-to-utc property is missing
2. convert Iceberg timestamptz(_ns) to Avro timestamp-* without the adjust-to-utc property
3. convert Iceberg timestamp(_ns) to Avro local-timestamp-*?

Personally, I'm only interested in ingesting Avro encoded records from Kafka using Flink, similar to the official example. So, even if no one adds support for Avro native timestamps to DataReader, PlannedDataReader and DataWriter, the changes in this PR are still useful on their own.

[RussellSpitzer]

I'm not sure why we would keep it scoped to just this class? As is, we are creating a function which other libraries could use but would produce records or schema which wouldn't work properly when the rest of this library worked with them.

So for example if you wanted to change SchemaToType in isolation I think that's pretty safe. We never would create output that would break other parts of the library. The moment we touch TypeToSchema we are potentially producing records that would break the rest of the library.

So I think we either follow the path of #Unknown and TimestampNanos
Or restrict the changes here just to the read side. Personally I think just fixing everything is the right move here.

[kinolaev]

The moment we touch TypeToSchema we are potentially producing records that would break the rest of the library.

I see your point. However, I suspect there are many places where schemas produced with legacyTimestampMapping=false will require additional work on the data side. For example, if this PR makes it to the main branch, I'd like to propose the corresponding changes to the Flink AvroGenericRecordToRowDataMapper and RowDataToAvroGenericRecordConverter classes (#17200).

Since the new behavior of TypeToSchema is flagged, support on the data side can be added gradually by people who are interested in following Avro timestamp semantics in specific parts of the Iceberg codebase. However, any data conversions will require updated schemas first.

Or restrict the changes here just to the read side.

If this is an option, I'd like to go this route. Personally, I am only interested in the read side. I simply found that adding symmetric support to the write side was quite easy and might be useful for others (given that there are several similar threads). That is the only reason why I included the changes to TypeToSchema to this PR and to RowDataToAvroGenericRecordConverter to PR #17200.

Otherwise, I will need some help with the data side. @AnatolyPopov , is there a chance you could rebase your PR #15437 on my branch and add support for the changes in DataWriter?

DataReader.java:163-164
PlannedDataReader.java:171-172
DataWriter.java:140-141

@RussellSpitzer , are there other places where you'd like to have Avro native timestamp support?

[kinolaev]

My preference is to keep the converter semantically correct (withoutZone() always produces local-timestamp-micros) and push the legacy wire format requirement to the callers that actually need it:

// Caller that is producing a schema that we want to write into AVRO
manifestPartitionSchema = toLegacyTimestamps(convert(icebergPartitionSchema));

I added package-private AvroLegacyTimestamps.convert, but I realized all 5 of its call sites are mostly dead code. Specifically: the BaseFile and GenericManifestEntry constructors, GenericDataFile.getAvroSchema, GenericDeleteFile.getAvroSchema, and PartitionData.partitionDataSchema. They all generate an Avro schema solely to implement the IndexedRecord interface, which is only used by the kafka-connect module via AvroEncoderUtil.encode->GenericAvroWriter to serialize and deserialize events in the control topic (except for GenericManifestEntry, whose IndexedRecord support isn't used anywhere). Should we keep the local to legacy timestamp conversion in these 5 classes?

We can't change the Avro encoding in manifests from what the spec currently says, and we can't change the spec for V3. (We could update V4 to accept or require the new types, but that's a separate discussion...)

The manifest writers aren't broken today because legacyTimestampMapping=true is the default. But that's the problem: the flag is in the wrong place. If we ever flip the default, or a caller adopts the new non-legacy path, the manifest writers silently produce spec-incompatible output with no indication at the call site that anything special is happening.

The actual schema is written via ManifestWriter.V{Version}(Delete)Writer->InternalData.write->Avro.write. That means it depends on the write.avro.local-timestamp.enabled property. I've pinned it to false for all versions (including v4) and removed local-timestamp-* from the spec for now. Once this PR merges into main, I plan to open a follow-up PR about supporting local-timestamp-* in the spec.

Thanks @RussellSpitzer for the review! Could you please take another look at the PR when you get a chance?

[kinolaev]

@RussellSpitzer, I'd be happy to keep working on this and hopefully get the PR wrapped up this week. Please take another look when you get a chance.

I've rebased my branch to resolve conflicts. There are no other changes since my last comment.

Status:

1. AvroSchemaUtil:
   1. Iceberg->Avro: local-timestamp-* by default, no flag
   2. Avro->Iceberg: adjustToUtcDefault flag, false by default
   3. isTimestamptz is inlined at its call sites
2. Writers and readers:
   1. support local-timestamp-*
   2. adjustToUtcDefault can be configured via the SupportsLocalTimestamp interface
3. Avro passes the write.avro.local-timestamp.enabled (default is false) and read.avro.adjust-to-utc.default (default is false) properties to the writers and readers
4. Manifests:
   1. adjustToUtcDefault is pinned to false in Avro->Iceberg convert calls
   2. package-private AvroLegacyTimestamps is introduced to convert local-timestamp-* back to timestamp-* across 5 classes for backward compatibility (see my previous comment)
   3. write.avro.local-timestamp.enabled is pinned to false for all manifest writer versions

upd: I saw on the mailing list that the 1.12.0 RC release is planned for this week. With your help, I'd like to focus on this PR so we can get it merged into main before the release.

upd2: Do you want to keep Iceberg convert(Avro) and Iceberg toIceberg(Avro) overloads without adjustToUtcDefault? It seems like they are only used in tests now.

[kinolaev]

I rebased the branch to resolve a conflict - PositionAndRowDatumWriter was recently removed.

I started thinking about a proposal for v4. My personal preference is to get rid of all inconsistencies between the specs at once:

1. timestamp(_ns) <-> local-timestamp-*
2. timestamptz(_ns) <-> timestamp-*
3. drop the adjust-to-utc property from v4

I realized that we don't actually need custom table properties. On the write side, we can just use format-version < 4 ? legacy iceberg timestamps : avro timestamps. On the read side, we need a property that tells us at which format-version a manifest or a data file was written. Manifests already have a required format-version metadata field that we can use to interpret the identity/void(timestamp(_ns)) partition field schema. And I think we could add a file_format_version field to manifest_entry, which should be null unless it differs from its manifest's format-version metadata field. Then, on the read side, we could do the following:

int fileFormatVersion = entry.fileFormatVersion() != null
    ? entry.fileFormatVersion()
    : manifest.formatVersion() != null ? manifest.formatVersion() : 1;
if (fileFormatVersion < 4) {
  // read `adjust-to-utc` with default `false`
} else {
  // follow the Avro spec, ignore `adjust-to-utc`
}

As a bonus, it allows us to add external Avro data files with file_format_version=3 or file_format_version=4 depending on their schema. And of course it leaves room for further file format changes.

I'm still exploring alternatives. Just wanted to share the best idea I have for now, in case you'd like to give early feedback - I'd really appreciate it! :)