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]

@AnatolyPopov @RussellSpitzer just following up on the changes I posted last week in the thread above: #17196 (comment). Whenever you have time for a review, I'm ready to keep pushing this PR forward.

[kinolaev]

@AnatolyPopov @RussellSpitzer I'm still hoping to get another review from you and I'm ready to continue working on this PR.

Current state:

• AvroSchemaUtil Iceberg->Avro conversion methods use local-timestamp-* and timestamp-* with adjust-to-utc=true by default.
• AvroSchemaUtil Avro->Iceberg conversion methods read timestamp-* without adjust-to-utc as zoned by default.
• Data writers support local-timestamp-* and expect zoned timestamps for timestamp-* without adjust-to-utc by default.
• Data readers support local-timestamp-* and read timestamp-* without adjust-to-utc as zoned by default.
• Data writers and readers can be configured using the SupportsLocalTimestamp interface.
• The write.avro.local-timestamp.enabled table property (documented, false by default) is passed to a data writer through SupportsLocalTimestamp, so default behavior remains unchanged.
• Readers accept the read.avro.adjust-to-utc.default table property (undocumented, false by default) through SupportsLocalTimestamp, but unfortunately no engine currently passes table properties to the readers.

Let me know if you would like to see the last piece - wiring table properties to the readers - included in this PR.

[kinolaev]

@AnatolyPopov @RussellSpitzer , I'm still ready to continue working on this PR and would appreciate another review from you. If your bandwidth is full, could you please ask someone else for a review?

[RussellSpitzer]

I still don't think we want to add all these public methods. All the Convert(Iceberg Thing) -> Avro Thing should not have a legacy flag. There is no ambiguity here and we don't need to emulate the old behavior by producing an incorrect type just because we used to.

For the other direction convert(Avro Thing) -> Iceberg Thing. We have ambiguity based on what the old writer of the avro schema produced. Only those are ambiguous and we need the ability to handle them in either direction.

convert(Schema)                          ← no-arg, defaults adjustToUtcDefault=true
convert(Schema, boolean)                 ← explicit adjustToUtcDefault
toIceberg(Schema)                        ← no-arg wrapper, returns iceberg.Schema
toIceberg(Schema, boolean)               ← explicit adjustToUtcDefault

The no-arg options here should default to adjustUtcDefault = False to match AVRO_ADJUST_TO_UTC_DEFAULT_DEFAULT.

There is also one other method
isTimestamptz(Schema, boolean) which is public and i'm not sure it needs to be. It should probably be inlined or something like to keep it out of our public api.

[kinolaev]

Thanks @RussellSpitzer!

Does this mean you want to get rid of the write.avro.local-timestamp.enabled table property and make Avro.write encode timestamp(_ns) as local-timestamp-* unconditionally?

There is an edge case covered in core/src/test/java/org/apache/iceberg/TestTimestampPartitions.java: when a partition spec contains an identity(timestamp) field, according to the spec, it should be encoded in manifests as timestamp-micros with adjust-to-utc=false. Do you want me to update the spec to allow encoding timestamp(_ns) as local-timestamp-* in manifests?

convert(Schema) ← no-arg, defaults adjustToUtcDefault=true

The no-arg options here should default to adjustUtcDefault = False to match AVRO_ADJUST_TO_UTC_DEFAULT_DEFAULT.

Could you please clarify your position on the default value for adjustUtcDefault? I'd prefer to keep the Avro-compliant default (true) in AvroSchemaUtil, as it is not specific to Iceberg, and can be used with externally produced schemas. I don't think it should match AVRO_ADJUST_TO_UTC_DEFAULT_DEFAULT, because the latter is specific to Iceberg, and changing it would be a breaking change.

[kinolaev]

@RussellSpitzer, I've reread our discussion.

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

Based on this suggestion, I dropped legacy Iceberg->Avro public methods, but kept the write.avro.local-timestamp.enabled table property.

AvroSchemaUtil.isTimestamptz has been inlined into (Planned)DataReader, DataWriter, and SchemaToType, the method was removed. String values "true"/"false" are no longer supported (they never were spec-compliant). If you want to keep string support, I think it would be better to retain AvroSchemaUtil.isTimestamptz (or add AvroSchemaUtil.getBooleanProp(String name, boolean defaultValue) instead).

Two questions remain:

1. Partition fields identity(timestamp)/identity(timestamp_ns)/void(timestamp)/void(timestamp_ns) are now encoded as local-timestamp-* in manifests, which breaks old readers - is that acceptable? For now, I've added local-timestamp-* to the spec to reflect the change. Fortunately, there are no other timestamp/timestamp_ns types in the manifest schemas for now.
2. adjustToUtcDefault: it is now Avro-compliant (true) in AvroSchemaUtil, DataWriter and readers (configurable via SupportsLocalTimestamp), and Iceberg-compliant (false) in manifests and for the read.avro.adjust-to-utc.default table property. Do you prefer to have it Iceberg-compliant (false) everywhere?

[kinolaev]

@RussellSpitzer, I reread your last message and realized that:

convert(Schema) ← no-arg, defaults adjustToUtcDefault=true

most probably was a description of the branch state at that time, rather than a suggestion to keep true by default. That is why I updated the PR again, and I think it should now be very close to what you envisioned:

1. AvroSchemaUtil Iceberg->Avro conversion methods produce local-timestamp-* unconditionally, which is allowed by the updated spec
2. The write.avro.local-timestamp.enabled table property (false by default) controls Avro data file schemas
3. The adjust-to-utc default is now false everywhere (matches the spec)
4. AvroSchemaUtil Avro->Iceberg conversion methods have overloads to allow setting adjustToUtcDefault=true
5. DataWriter and readers allow setting adjustToUtcDefault=true via the SupportsLocalTimestamp interface
6. isTimestamptz has been inlined

The main question right now is: is it acceptable to encode partition fields identity(timestamp), identity(timestamp_ns), void(timestamp) and void(timestamp_ns) as local-timestamp-* in manifests?

A smaller question: is it OK to throw an exception when the adjust-to-utc value is not a boolean?

PS: I found several more places in Flink where legacyTimestampMapping was still true by default and flipped them to false (d1c8e61).

[RussellSpitzer]

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 and it's also somewhat moot since V4 manifests are no longer constrained to Avro, so that encoding question may not arise there.)

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.

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));

That way the exception is visible and explicit where it matters, rather than a hidden mode in the utility. The utility should only take a mode flag when the output is genuinely ambiguous from the input alone and local-timestamp-* is unambiguous.

If you want to throw an exception when adjust-to-utc is not a boolean that sounds fine to me. Generally we would only do this at the moment of conversion though so make sure it's not a parse error but an error when we are trying to do something with the type that is undefined.