[issue-7357] Add support for Avro logical types in both realtime and batch ingestion by ddcprg · Pull Request #7358 · apache/pinot

ddcprg · 2021-08-24T18:07:38Z

Description

Fixes #7357

I'm looking to get some feedback before making this a full PR

Upgrade Notes

Does this PR prevent a zero down-time upgrade? (Assume upgrade order: Controller, Broker, Server, Minion)

Yes (Please label as backward-incompat, and complete the section below on Release Notes)

Does this PR fix a zero-downtime upgrade introduced earlier?

Yes (Please label this as backward-incompat, and complete the section below on Release Notes)

Does this PR otherwise need attention when creating release notes? Things to consider:

New configuration options
Deprecation of configurations
Signature changes to public methods/interfaces
New plugins added or old plugins removed

Yes (Please label this PR as release-notes and complete the section on Release Notes)

The code change should be backwards compatible.

Release Notes

Documentation

ddcprg · 2021-09-07T19:29:28Z

I'm looking at modifying the data used by RealtimeClusterIntegrationTest to add fields with logical types. Any guidelines on how this data has been generated?

codecov-commenter · 2021-09-28T23:06:59Z

Codecov Report

Merging #7358 (c5eed6e) into master (ec42a3a) will decrease coverage by 33.55%.
The diff coverage is 0.00%.

@@              Coverage Diff              @@
##             master    #7358       +/-   ##
=============================================
- Coverage     70.35%   36.79%   -33.56%     
+ Complexity     4406       84     -4322     
=============================================
  Files          1690     1693        +3     
  Lines         88187    88247       +60     
  Branches      13190    13198        +8     
=============================================
- Hits          62042    32470    -29572     
- Misses        21856    53163    +31307     
+ Partials       4289     2614     -1675

Flag	Coverage Δ
integration1	`27.50% <0.00%> (+0.05%)`	⬆️
integration2	`25.50% <0.00%> (-0.17%)`	⬇️
unittests1	`?`
unittests2	`14.29% <0.00%> (-0.01%)`	⬇️

Flags with carried forward coverage won't be shown. Click here to find out more.

Impacted Files	Coverage Δ
...t/plugin/inputformat/avro/AvroRecordExtractor.java	`0.00% <0.00%> (-89.48%)`	⬇️
...in/inputformat/avro/AvroRecordExtractorConfig.java	`0.00% <0.00%> (ø)`
...inot/plugin/inputformat/avro/AvroRecordReader.java	`0.00% <0.00%> (-88.89%)`	⬇️
...lugin/inputformat/avro/AvroRecordReaderConfig.java	`0.00% <0.00%> (ø)`
.../pinot/plugin/inputformat/avro/AvroSchemaUtil.java	`0.00% <0.00%> (-18.19%)`	⬇️
...ugin/inputformat/avro/KafkaAvroMessageDecoder.java	`0.00% <0.00%> (ø)`
...gin/inputformat/avro/SimpleAvroMessageDecoder.java	`0.00% <0.00%> (ø)`
.../pinot/spi/data/readers/RecordExtractorConfig.java	`0.00% <0.00%> (ø)`
...he/pinot/spi/data/readers/RecordReaderFactory.java	`0.00% <0.00%> (-57.50%)`	⬇️
...ava/org/apache/pinot/spi/plugin/PluginManager.java	`0.00% <0.00%> (-57.97%)`	⬇️
... and 947 more

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update ec42a3a...c5eed6e. Read the comment docs.

ddcprg · 2021-09-29T13:54:49Z

I thought changing the commit hash would re-trigger the build

ddcprg · 2021-10-07T16:58:27Z

...ot-avro-base/src/main/java/org/apache/pinot/plugin/inputformat/avro/AvroRecordExtractor.java

Note that BigDecimal is converted to Double as Pinot does not have support for big number columns at the moment and introducing a new type is out of scope in this ticket

once bigdecimal type is fully supported a record reader option can control this behaviour

...rc/main/java/org/apache/pinot/plugin/inputformat/avro/AvroLogicalTypeConversionRegistry.java

richardstartin

👍🏻

walterddr

the approach looks good. thanks @ddcprg for the contribution. I left one minor comment otherwise good to go.

walterddr · 2022-03-03T18:39:17Z

...rc/main/java/org/apache/pinot/plugin/inputformat/avro/AvroLogicalTypeConversionRegistry.java

can we put these in ArvoSchemUtils? is there a reason we need the Registry?

I'll rebase, review the code since it's been a while and get back to you with the reason why I put this here

there is not reason to not to move it ArvoSchemUtils other than having some separation of concern,, I'll move them over if you think that's a better place for this

this looks good now.

ddcprg · 2022-04-06T09:49:47Z

...ot-avro-base/src/main/java/org/apache/pinot/plugin/inputformat/avro/AvroRecordExtractor.java

this setting should come from table config in order to avoid potentially breaking existing ingestion that come with logical types, however RecordExtractorConfig is always null. Is anyone working on this logic at the moment? I could give that a go as I'll need to add the setting for this feature

yes I think we can create a AvroRecordExtractorConfig and provide some defaults to this if necessary. see CsvRecordExtractorConfig as an example. i think it can be part of this PR

ddcprg · 2022-04-06T20:13:59Z

...avro-base/src/main/java/org/apache/pinot/plugin/inputformat/avro/AvroRecordReaderConfig.java

For ingestion jobs these properties can be configured via recordsReaderSpec. However I can't see how these properties can be configured for a realtime table - perhaps via ingestionSpec in the table definition? I was looking through the code but I couldn't figure it out. I believe this is not possible at the moment

I think I've found where the properties can be read from. Stream config has decoder properties (it seems like these are available in table config of realtime tables) which are propagated to the Avro decoder classes, it looks like I can take advantage of that to populate the logical type flag. I'll add a commit to the PR and let me know if you see anything missing @walterddr

hmm. not sure which one are you referring to. there's a RecordReaderConfig base class which provides configuration properties to RecordReaders (see RecordReaderFactory class). and it is apparently passed in via a readerConfigFile path (see. CreateSegmentCommand entrypoint)

Jackie-Jiang

LGTM otherwise, no major comments

Does the change also apply to batch ingestion flow? If so, let's update the PR name and description to reflect that

Jackie-Jiang · 2022-04-11T19:02:53Z

...o-base/src/main/java/org/apache/pinot/plugin/inputformat/avro/AvroRecordExtractorConfig.java

Suggested change

_enableLogicalTypes = Boolean.valueOf(props.getOrDefault("enableLogicalTypes", Boolean.FALSE.toString()));

_enableLogicalTypes = Boolean.parseBoolean(props.get("enableLogicalTypes"));

Jackie-Jiang · 2022-04-11T19:08:03Z

...ot-avro-base/src/main/java/org/apache/pinot/plugin/inputformat/avro/AvroRecordExtractor.java

(minor) Can be simplified

Suggested change

if (value instanceof BigDecimal) {

return Double.valueOf(((BigDecimal) value).doubleValue());

}

if (value instanceof LocalDate) {

return Long.valueOf(((LocalDate) value).toEpochDay());

}

if (value instanceof Instant) {

return Long.valueOf(((Instant) value).toEpochMilli());

}

if (value instanceof BigDecimal) {

return ((BigDecimal) value).doubleValue();

}

if (value instanceof LocalDate) {

return ((LocalDate) value).toEpochDay();

}

if (value instanceof Instant) {

return ((Instant) value).toEpochMilli();

}

Jackie-Jiang · 2022-04-11T19:08:39Z

...ot-avro-base/src/main/java/org/apache/pinot/plugin/inputformat/avro/AvroRecordExtractor.java

(format) Is this auto-formatted? We usually keep the javadoc multi lines

not sure, perhaps an old pinot code style I had in my IDE

Jackie-Jiang · 2022-04-11T19:09:31Z

...ot-avro-base/src/main/java/org/apache/pinot/plugin/inputformat/avro/AvroRecordExtractor.java

(format) This change does not follow the Pinot Style

Jackie-Jiang · 2022-04-11T19:14:04Z

...t/pinot-avro-base/src/main/java/org/apache/pinot/plugin/inputformat/avro/AvroSchemaUtil.java

(format) We usually keep the javadoc multi lines

Jackie-Jiang · 2022-04-11T19:14:10Z

...t/pinot-avro-base/src/main/java/org/apache/pinot/plugin/inputformat/avro/AvroSchemaUtil.java

(format) We usually keep the javadoc multi lines

Jackie-Jiang · 2022-04-11T19:14:57Z

...t-avro-base/src/test/java/org/apache/pinot/plugin/inputformat/avro/AvroRecordReaderTest.java

We don't need to change this (also want to test the behavior of passing null)

Jackie-Jiang · 2022-04-11T19:15:08Z

...rc/test/java/org/apache/pinot/plugin/inputformat/avro/AvroRecordToPinotRowGeneratorTest.java

We don't need to change this (also want to test the behavior of passing null)

Jackie-Jiang · 2022-04-11T19:16:41Z

.../pinot/plugin/inputformat/avro/confluent/KafkaConfluentSchemaRegistryAvroMessageDecoder.java

Suggested change

AvroRecordExtractorConfig config = PluginManager.get().createInstance(AvroRecordExtractorConfig.class.getName());

AvroRecordExtractorConfig config = new AvroRecordExtractorConfig();

Jackie-Jiang · 2022-04-11T19:17:48Z

pinot-spi/src/main/java/org/apache/pinot/spi/data/readers/RecordExtractorConfig.java

(convention) We don't usually annotate non-null, but only annotate nullable. Also make it multiple lines

Suggested change

default void init(@Nonnull Map<String, String> props) { }

default void init(Map<String, String> props) {

}

ddcprg · 2022-04-13T07:27:40Z

@Jackie-Jiang thanks for reviewing, the PR covers both batch and realtime ingestion - I've renamed the PR. There is one more thing I'd like to do on this PR before I merge it. I'm looking at adding a full integration test (if possible) for both batch and ingestion, if you have any pointer please let me know. I'll also tidy up the commits

Jackie-Jiang · 2022-04-14T00:18:17Z

@ddcprg You may take a look at HybridClusterIntegrationTest on how to feed the data into both batch flow and streaming flow. In order to add an integration test, we need to add a new avro data set with the logical types. IMO it might be too much for this support (adding an integration test can add 1-2 minutes to the CI). A thorough unit test on avro record extractor should be good enough.

ddcprg · 2022-04-14T08:50:02Z

Thanks @Jackie-Jiang in that case I'll add more unit tests around record extractor. I'll squash all the commits into one after that

ddcprg · 2022-04-14T10:59:09Z

...ot-avro-base/src/main/java/org/apache/pinot/plugin/inputformat/avro/AvroRecordExtractor.java

I'm thinking it may be better to initially return a String representation of the timestamp instead, Pinot can coerce the value into a timestamp. At this point I can't make a guess of whether the user has specified millis or micros. Behaviour control could be configured via record reader settings in the future. @Jackie-Jiang @walterddr let me know what you think

Seems the Instant is always converted from the timestamp type, so we might want to return the Timestamp object here.

apologies for the delay, this now returns java.sql.Timestamp instead

Jackie-Jiang · 2022-04-14T21:23:15Z

...ot-avro-base/src/main/java/org/apache/pinot/plugin/inputformat/avro/AvroRecordExtractor.java

(minor) revert

Jackie-Jiang · 2022-04-14T21:33:48Z

...t/pinot-avro-base/src/main/java/org/apache/pinot/plugin/inputformat/avro/AvroSchemaUtil.java

(minor) Revert, same for other places (is this auto-formatted?)

yes, not sure why my IDE is doing this

Jackie-Jiang · 2022-04-14T21:35:22Z

...t/pinot-avro-base/src/main/java/org/apache/pinot/plugin/inputformat/avro/AvroSchemaUtil.java

(minor) Let's put this static block next to the map construction for readability

Jackie-Jiang · 2022-04-14T21:42:43Z

...ot-avro-base/src/main/java/org/apache/pinot/plugin/inputformat/avro/AvroRecordExtractor.java

Seems the Instant is always converted from the timestamp type, so we might want to return the Timestamp object here.

Jackie-Jiang · 2022-05-09T18:10:49Z

...ot-avro-base/src/main/java/org/apache/pinot/plugin/inputformat/avro/AvroRecordExtractor.java

We recently added big-decimal support in #8503, so here we should just return BigDecimal as is (no need to have this special handling. The base implementation can handle it properly)

this is great! I've now changed this line and squashed all the commits into a single one

ddcprg force-pushed the avro_logical_types branch from 133d80b to bbb220b Compare September 7, 2021 19:25

ddcprg changed the title ~~WIP [issue-7357] Add support for Avro logical types in realtime ingestion~~ [issue-7357] Add support for Avro logical types in realtime ingestion Sep 7, 2021

ddcprg marked this pull request as ready for review September 7, 2021 19:31

ddcprg force-pushed the avro_logical_types branch from b1122c7 to 183b5b1 Compare September 29, 2021 13:52

ddcprg commented Oct 7, 2021

View reviewed changes

richardstartin reviewed Mar 3, 2022

View reviewed changes

...rc/main/java/org/apache/pinot/plugin/inputformat/avro/AvroLogicalTypeConversionRegistry.java Outdated Show resolved Hide resolved

richardstartin approved these changes Mar 3, 2022

View reviewed changes

walterddr reviewed Mar 3, 2022

View reviewed changes

ddcprg force-pushed the avro_logical_types branch 2 times, most recently from 2c5f819 to c159f0f Compare April 6, 2022 09:27

ddcprg commented Apr 6, 2022

View reviewed changes

Jackie-Jiang approved these changes Apr 11, 2022

View reviewed changes

ddcprg changed the title ~~[issue-7357] Add support for Avro logical types in realtime ingestion~~ [issue-7357] Add support for Avro logical types in both realtime and batch ingestion Apr 13, 2022

ddcprg force-pushed the avro_logical_types branch from e56ab9c to fd5ed95 Compare April 13, 2022 07:31

ddcprg force-pushed the avro_logical_types branch 2 times, most recently from b5ebdf9 to 7dd63fd Compare April 14, 2022 10:47

ddcprg commented Apr 14, 2022

View reviewed changes

Jackie-Jiang reviewed Apr 14, 2022

View reviewed changes

ddcprg force-pushed the avro_logical_types branch 2 times, most recently from b69a11b to 7c134c3 Compare May 9, 2022 12:08

Jackie-Jiang reviewed May 9, 2022

View reviewed changes

ddcprg force-pushed the avro_logical_types branch from 7c134c3 to e9b8911 Compare May 9, 2022 19:31

[issue-7357] Add support for Avro logical types

c5eed6e

ddcprg force-pushed the avro_logical_types branch from e9b8911 to c5eed6e Compare May 9, 2022 19:35

Jackie-Jiang merged commit b494345 into apache:master May 9, 2022

	_enableLogicalTypes = Boolean.valueOf(props.getOrDefault("enableLogicalTypes", Boolean.FALSE.toString()));
	_enableLogicalTypes = Boolean.parseBoolean(props.get("enableLogicalTypes"));

-    if (value instanceof BigDecimal) {
-      return Double.valueOf(((BigDecimal) value).doubleValue());
-    }
-    if (value instanceof LocalDate) {
-      return Long.valueOf(((LocalDate) value).toEpochDay());
-    }
-    if (value instanceof Instant) {
-      return Long.valueOf(((Instant) value).toEpochMilli());
-    }
+    if (value instanceof BigDecimal) {
+      return ((BigDecimal) value).doubleValue();
+    }
+    if (value instanceof LocalDate) {
+      return ((LocalDate) value).toEpochDay();
+    }
+    if (value instanceof Instant) {
+      return ((Instant) value).toEpochMilli();
+    }

	AvroRecordExtractorConfig config = PluginManager.get().createInstance(AvroRecordExtractorConfig.class.getName());
	AvroRecordExtractorConfig config = new AvroRecordExtractorConfig();

	default void init(@Nonnull Map<String, String> props) { }
	default void init(Map<String, String> props) {
	}

Conversation

ddcprg commented Aug 24, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Description

Upgrade Notes

Release Notes

Documentation

Uh oh!

ddcprg commented Sep 7, 2021

Uh oh!

codecov-commenter commented Sep 28, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

ddcprg commented Sep 29, 2021

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

richardstartin left a comment

Choose a reason for hiding this comment

Uh oh!

walterddr left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ddcprg Apr 6, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ddcprg Apr 6, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ddcprg Apr 8, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Jackie-Jiang left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ddcprg commented Apr 13, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Jackie-Jiang commented Apr 14, 2022

ddcprg commented Aug 24, 2021 •

edited

Loading

codecov-commenter commented Sep 28, 2021 •

edited

Loading

ddcprg Apr 6, 2022 •

edited

Loading

ddcprg Apr 6, 2022 •

edited

Loading

ddcprg Apr 8, 2022 •

edited

Loading

ddcprg commented Apr 13, 2022 •

edited

Loading

ddcprg May 9, 2022 •

edited

Loading