feat(schema): Remove direct usage of Avro schema in Flink-client path #17739
Conversation
| } | ||
|
|
||
| @Test | ||
| void testUnionSchemaWithMultipleRecordTypes() { |
There was a problem hiding this comment.
These are copied from the TestAvroSchemaConverter
github-actions
bot
added
size:XL
| return rowDataQueryContext.getFieldQueryContext(column).getValAsJava(data, allowsNull); | ||
| } | ||
|
|
||
| private Object getColumnValue(Schema recordSchema, String column, Properties props) { |
There was a problem hiding this comment.
Can we just change the param to take in HoodieSchema so that then in line 204 we dont have to do fromAvroSchema, or does this result in large changes?
There was a problem hiding this comment.
This is a private method so it is fine but as mentioned in the description, the record classes are handled in a separate ticket
| return getColumnValueAsJava(recordSchema, column, props, true); | ||
| } | ||
|
|
||
| private Object getColumnValueAsJava(Schema recordSchema, String column, Properties props, boolean allowsNull) { |
There was a problem hiding this comment.
Same as comment made here https://github.com/apache/hudi/pull/17739/files#r2651367282
| } | ||
|
|
||
| @Override | ||
| public Option<HoodieAvroIndexedRecord> toIndexedRecord(Schema recordSchema, Properties props) { |
There was a problem hiding this comment.
Same as comment https://github.com/apache/hudi/pull/17739/files#r2651367282
There was a problem hiding this comment.
As mentioned in the PR description, the Record interface will not be updated as part of this.
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableFactory.java
Outdated
Show resolved
Hide resolved
|
@danny0405 can you give this a review? |
the-other-tim-brown
force-pushed
the
flink-client-schema-migration
branch
from
f57f68d to
02ba0c6
Compare
the-other-tim-brown
force-pushed
the
flink-client-schema-migration
branch
from
02ba0c6 to
6bc7f5c
Compare
| import java.util.Arrays; | ||
|
|
||
| import static org.hamcrest.CoreMatchers.is; | ||
| import static org.hamcrest.MatcherAssert.assertThat; |
There was a problem hiding this comment.
Super Nit:
While there are some parts of our code that are using hamcrest for assertThat, I feel assertj is more widely used. e.g. Trino and flink-cdc chose it over hamcrest.
import static org.assertj.core.api.Assertions.assertThat;We have both in our code base now, i feel we should clean this up in a chore PR moving forward.
| @Override | ||
| public Object getColumnValueAsJava(Schema recordSchema, String column, Properties props) { | ||
| return getColumnValueAsJava(recordSchema, column, props, true); | ||
| return getColumnValueAsJava(HoodieSchema.fromAvroSchema(recordSchema), column, props, true); |
There was a problem hiding this comment.
do we have plan to replace the param as a HoodieSchema too: getColumnValueAsJava(HoodieSchema recordSchema to avoid the per-record schema creation overhead.
| boolean utcTimezone = Boolean.parseBoolean(props.getProperty( | ||
| HoodieStorageConfig.WRITE_UTC_TIMEZONE.key(), HoodieStorageConfig.WRITE_UTC_TIMEZONE.defaultValue().toString())); | ||
| RowDataQueryContext rowDataQueryContext = RowDataAvroQueryContexts.fromAvroSchema(recordSchema, utcTimezone); | ||
| RowDataQueryContext rowDataQueryContext = RowDataQueryContexts.fromSchema(recordSchema, utcTimezone); |
There was a problem hiding this comment.
confirmed that the cache should still work.
5 participants
Describe the issue this Pull Request addresses
Part 1 of #14280
This aims to remove direct usage of the avro schema in the flink client with the exception of the Flink Record objects which will be covered in #17689
Summary and Changelog
AvroSchemaConverterand updates all calls to go throughHoodieSchemaConverterRowDataToAvroConvertersconverter interface to useHoodieSchemaRowDataAvroQueryContextsto useHoodieSchema. Class name is updated toRowDataQueryContextsImpact
Updates the client to use our new schema system to allow us to add new types not available in Avro.
Risk Level
Low
Documentation Update
Contributor's checklist