You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by GitBox <gi...@apache.org> on 2021/03/01 08:11:05 UTC
[GitHub] [hudi] danny0405 opened a new pull request #2613: [HUDI-1647] Supports snapshot read for Flink
danny0405 opened a new pull request #2613:
URL: https://github.com/apache/hudi/pull/2613
## What is the purpose of the pull request
* COW: the parquet files for the latest file group slices
* MOR: the parquet base file + log files for the latest file group slices
Also implements the SQL connectors for both slink and source.
## Brief change log
- Add input formats for both COR(parquet) and MOR (parquet + log)
- Add table factory and table source/sink for flink
## Verify this pull request
Added UTs and ITs.
## Committer checklist
- [ ] Has a corresponding JIRA in PR title & commit
- [ ] Commit message is descriptive of the change
- [ ] CI is green
- [ ] Necessary doc changes done or have another open PR
- [ ] For large changes, please consider breaking it into sub-tasks under an umbrella JIRA.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] danny0405 commented on a change in pull request #2613: [HUDI-1647] Supports snapshot read for Flink
Posted by GitBox <gi...@apache.org>.
danny0405 commented on a change in pull request #2613:
URL: https://github.com/apache/hudi/pull/2613#discussion_r586099997
##########
File path: hudi-flink/src/main/java/org/apache/hudi/source/format/cow/AbstractColumnReader.java
##########
@@ -0,0 +1,324 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.source.format.cow;
+
+import org.apache.flink.formats.parquet.vector.ParquetDictionary;
+import org.apache.flink.formats.parquet.vector.reader.ColumnReader;
+import org.apache.flink.table.data.vector.writable.WritableColumnVector;
+import org.apache.flink.table.data.vector.writable.WritableIntVector;
+import org.apache.parquet.Preconditions;
+import org.apache.parquet.bytes.ByteBufferInputStream;
+import org.apache.parquet.bytes.BytesInput;
+import org.apache.parquet.bytes.BytesUtils;
+import org.apache.parquet.column.ColumnDescriptor;
+import org.apache.parquet.column.Dictionary;
+import org.apache.parquet.column.Encoding;
+import org.apache.parquet.column.page.DataPage;
+import org.apache.parquet.column.page.DataPageV1;
+import org.apache.parquet.column.page.DataPageV2;
+import org.apache.parquet.column.page.DictionaryPage;
+import org.apache.parquet.column.page.PageReader;
+import org.apache.parquet.column.values.ValuesReader;
+import org.apache.parquet.io.ParquetDecodingException;
+import org.apache.parquet.schema.PrimitiveType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+
+import static org.apache.parquet.column.ValuesType.REPETITION_LEVEL;
+
+/**
+ * Abstract {@link ColumnReader}.
+ * See {@link org.apache.parquet.column.impl.ColumnReaderImpl},
+ * part of the code is referred from Apache Spark and Apache Parquet.
+ *
+ * <p>Note: Reference Flink release 1.11.2 {@link org.apache.flink.formats.parquet.vector.reader.AbstractColumnReader}
+ * because some of the package scope methods.
+ */
+public abstract class AbstractColumnReader<V extends WritableColumnVector>
Review comment:
Flink have its own.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] codecov-io edited a comment on pull request #2613: [HUDI-1647] Supports snapshot read for Flink
Posted by GitBox <gi...@apache.org>.
codecov-io edited a comment on pull request #2613:
URL: https://github.com/apache/hudi/pull/2613#issuecomment-787825622
# [Codecov](https://codecov.io/gh/apache/hudi/pull/2613?src=pr&el=h1) Report
> Merging [#2613](https://codecov.io/gh/apache/hudi/pull/2613?src=pr&el=desc) (0d77db7) into [master](https://codecov.io/gh/apache/hudi/commit/7a11de12764d8f68f296c6e68a22822318bfbefa?el=desc) (7a11de1) will **decrease** coverage by `0.15%`.
> The diff coverage is `47.23%`.
[](https://codecov.io/gh/apache/hudi/pull/2613?src=pr&el=tree)
```diff
@@ Coverage Diff @@
## master #2613 +/- ##
============================================
- Coverage 51.56% 51.41% -0.16%
- Complexity 3286 3480 +194
============================================
Files 445 461 +16
Lines 20328 21678 +1350
Branches 2102 2299 +197
============================================
+ Hits 10483 11146 +663
- Misses 8978 9574 +596
- Partials 867 958 +91
```
| Flag | Coverage Δ | Complexity Δ | |
|---|---|---|---|
| hudicli | `36.87% <ø> (ø)` | `0.00 <ø> (ø)` | |
| hudiclient | `100.00% <ø> (ø)` | `0.00 <ø> (ø)` | |
| hudicommon | `51.32% <ø> (-0.01%)` | `0.00 <ø> (ø)` | |
| hudiflink | `50.31% <47.23%> (-1.09%)` | `0.00 <188.00> (ø)` | |
| hudihadoopmr | `33.16% <ø> (ø)` | `0.00 <ø> (ø)` | |
| hudisparkdatasource | `69.71% <ø> (ø)` | `0.00 <ø> (ø)` | |
| hudisync | `49.62% <ø> (ø)` | `0.00 <ø> (ø)` | |
| huditimelineservice | `66.49% <ø> (ø)` | `0.00 <ø> (ø)` | |
| hudiutilities | `69.59% <ø> (+0.05%)` | `0.00 <ø> (ø)` | |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags#carryforward-flags-in-the-pull-request-comment) to find out more.
| [Impacted Files](https://codecov.io/gh/apache/hudi/pull/2613?src=pr&el=tree) | Coverage Δ | Complexity Δ | |
|---|---|---|---|
| [...va/org/apache/hudi/factory/HoodieTableFactory.java](https://codecov.io/gh/apache/hudi/pull/2613/diff?src=pr&el=tree#diff-aHVkaS1mbGluay9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvaHVkaS9mYWN0b3J5L0hvb2RpZVRhYmxlRmFjdG9yeS5qYXZh) | `0.00% <0.00%> (ø)` | `0.00 <0.00> (?)` | |
| [.../org/apache/hudi/operator/StreamWriteFunction.java](https://codecov.io/gh/apache/hudi/pull/2613/diff?src=pr&el=tree#diff-aHVkaS1mbGluay9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvaHVkaS9vcGVyYXRvci9TdHJlYW1Xcml0ZUZ1bmN0aW9uLmphdmE=) | `84.00% <0.00%> (-2.60%)` | `22.00 <0.00> (ø)` | |
| [.../org/apache/hudi/operator/StreamWriteOperator.java](https://codecov.io/gh/apache/hudi/pull/2613/diff?src=pr&el=tree#diff-aHVkaS1mbGluay9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvaHVkaS9vcGVyYXRvci9TdHJlYW1Xcml0ZU9wZXJhdG9yLmphdmE=) | `0.00% <0.00%> (ø)` | `0.00 <0.00> (ø)` | |
| [...ache/hudi/operator/StreamWriteOperatorFactory.java](https://codecov.io/gh/apache/hudi/pull/2613/diff?src=pr&el=tree#diff-aHVkaS1mbGluay9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvaHVkaS9vcGVyYXRvci9TdHJlYW1Xcml0ZU9wZXJhdG9yRmFjdG9yeS5qYXZh) | `0.00% <0.00%> (ø)` | `0.00 <0.00> (ø)` | |
| [...udi/operator/partitioner/BucketAssignFunction.java](https://codecov.io/gh/apache/hudi/pull/2613/diff?src=pr&el=tree#diff-aHVkaS1mbGluay9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvaHVkaS9vcGVyYXRvci9wYXJ0aXRpb25lci9CdWNrZXRBc3NpZ25GdW5jdGlvbi5qYXZh) | `85.86% <ø> (+0.92%)` | `22.00 <0.00> (+1.00)` | |
| [...ain/java/org/apache/hudi/sink/HoodieTableSink.java](https://codecov.io/gh/apache/hudi/pull/2613/diff?src=pr&el=tree#diff-aHVkaS1mbGluay9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvaHVkaS9zaW5rL0hvb2RpZVRhYmxlU2luay5qYXZh) | `0.00% <0.00%> (ø)` | `0.00 <0.00> (?)` | |
| [...e/hudi/source/format/cow/ParquetDecimalVector.java](https://codecov.io/gh/apache/hudi/pull/2613/diff?src=pr&el=tree#diff-aHVkaS1mbGluay9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvaHVkaS9zb3VyY2UvZm9ybWF0L2Nvdy9QYXJxdWV0RGVjaW1hbFZlY3Rvci5qYXZh) | `0.00% <0.00%> (ø)` | `0.00 <0.00> (?)` | |
| [.../org/apache/hudi/util/AvroToRowDataConverters.java](https://codecov.io/gh/apache/hudi/pull/2613/diff?src=pr&el=tree#diff-aHVkaS1mbGluay9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvaHVkaS91dGlsL0F2cm9Ub1Jvd0RhdGFDb252ZXJ0ZXJzLmphdmE=) | `16.94% <16.94%> (ø)` | `12.00 <12.00> (?)` | |
| [...hudi/source/format/cow/ParquetSplitReaderUtil.java](https://codecov.io/gh/apache/hudi/pull/2613/diff?src=pr&el=tree#diff-aHVkaS1mbGluay9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvaHVkaS9zb3VyY2UvZm9ybWF0L2Nvdy9QYXJxdWV0U3BsaXRSZWFkZXJVdGlsLmphdmE=) | `20.00% <20.00%> (ø)` | `13.00 <13.00> (?)` | |
| [...java/org/apache/hudi/util/AvroSchemaConverter.java](https://codecov.io/gh/apache/hudi/pull/2613/diff?src=pr&el=tree#diff-aHVkaS1mbGluay9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvaHVkaS91dGlsL0F2cm9TY2hlbWFDb252ZXJ0ZXIuamF2YQ==) | `23.74% <25.97%> (+23.74%)` | `13.00 <7.00> (+13.00)` | |
| ... and [39 more](https://codecov.io/gh/apache/hudi/pull/2613/diff?src=pr&el=tree-more) | |
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] yanghua merged pull request #2613: [HUDI-1647] Supports snapshot read for Flink
Posted by GitBox <gi...@apache.org>.
yanghua merged pull request #2613:
URL: https://github.com/apache/hudi/pull/2613
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] yanghua commented on a change in pull request #2613: [HUDI-1647] Supports snapshot read for Flink
Posted by GitBox <gi...@apache.org>.
yanghua commented on a change in pull request #2613:
URL: https://github.com/apache/hudi/pull/2613#discussion_r585432482
##########
File path: hudi-flink/src/main/java/org/apache/hudi/source/format/cow/AbstractColumnReader.java
##########
@@ -0,0 +1,324 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.source.format.cow;
+
+import org.apache.flink.formats.parquet.vector.ParquetDictionary;
+import org.apache.flink.formats.parquet.vector.reader.ColumnReader;
+import org.apache.flink.table.data.vector.writable.WritableColumnVector;
+import org.apache.flink.table.data.vector.writable.WritableIntVector;
+import org.apache.parquet.Preconditions;
+import org.apache.parquet.bytes.ByteBufferInputStream;
+import org.apache.parquet.bytes.BytesInput;
+import org.apache.parquet.bytes.BytesUtils;
+import org.apache.parquet.column.ColumnDescriptor;
+import org.apache.parquet.column.Dictionary;
+import org.apache.parquet.column.Encoding;
+import org.apache.parquet.column.page.DataPage;
+import org.apache.parquet.column.page.DataPageV1;
+import org.apache.parquet.column.page.DataPageV2;
+import org.apache.parquet.column.page.DictionaryPage;
+import org.apache.parquet.column.page.PageReader;
+import org.apache.parquet.column.values.ValuesReader;
+import org.apache.parquet.io.ParquetDecodingException;
+import org.apache.parquet.schema.PrimitiveType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+
+import static org.apache.parquet.column.ValuesType.REPETITION_LEVEL;
+
+/**
+ * Abstract {@link ColumnReader}.
+ * See {@link org.apache.parquet.column.impl.ColumnReaderImpl},
+ * part of the code is referred from Apache Spark and Apache Parquet.
+ *
+ * <p>Note: Reference Flink release 1.11.2 {@link org.apache.flink.formats.parquet.vector.reader.AbstractColumnReader}
+ * because some of the package scope methods.
+ */
+public abstract class AbstractColumnReader<V extends WritableColumnVector>
+ implements ColumnReader<V> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(org.apache.flink.formats.parquet.vector.reader.AbstractColumnReader.class);
+
+ private final PageReader pageReader;
+
+ /**
+ * The dictionary, if this column has dictionary encoding.
+ */
+ protected final Dictionary dictionary;
+
+ /**
+ * Maximum definition level for this column.
+ */
+ protected final int maxDefLevel;
+
+ protected final ColumnDescriptor descriptor;
+
+ /**
+ * Total number of values read.
+ */
+ private long valuesRead;
+
+ /**
+ * value that indicates the end of the current page. That is, if valuesRead ==
+ * endOfPageValueCount, we are at the end of the page.
+ */
+ private long endOfPageValueCount;
+
+ /**
+ * If true, the current page is dictionary encoded.
+ */
+ private boolean isCurrentPageDictionaryEncoded;
+
+ /**
+ * Total values in the current page.
+ */
+ private int pageValueCount;
+
+ /*
+ * Input streams:
+ * 1.Run length encoder to encode every data, so we have run length stream to get
+ * run length information.
+ * 2.Data maybe is real data, maybe is dictionary ids which need be decode to real
+ * data from Dictionary.
+ *
+ * Run length stream ------> Data stream
+ * |
+ * ------> Dictionary ids stream
+ */
+
+ /**
+ * Run length decoder for data and dictionary.
+ */
+ protected RunLengthDecoder runLenDecoder;
+
+ /**
+ * Data input stream.
+ */
+ ByteBufferInputStream dataInputStream;
+
+ /**
+ * Dictionary decoder to wrap dictionary ids input stream.
+ */
+ private RunLengthDecoder dictionaryIdsDecoder;
+
+ public AbstractColumnReader(
+ ColumnDescriptor descriptor,
+ PageReader pageReader) throws IOException {
+ this.descriptor = descriptor;
+ this.pageReader = pageReader;
+ this.maxDefLevel = descriptor.getMaxDefinitionLevel();
+
+ DictionaryPage dictionaryPage = pageReader.readDictionaryPage();
+ if (dictionaryPage != null) {
+ try {
+ this.dictionary = dictionaryPage.getEncoding().initDictionary(descriptor, dictionaryPage);
+ this.isCurrentPageDictionaryEncoded = true;
+ } catch (IOException e) {
+ throw new IOException("could not decode the dictionary for " + descriptor, e);
+ }
+ } else {
+ this.dictionary = null;
+ this.isCurrentPageDictionaryEncoded = false;
+ }
+ /*
+ * Total number of values in this column (in this row group).
+ */
+ long totalValueCount = pageReader.getTotalValueCount();
+ if (totalValueCount == 0) {
+ throw new IOException("totalValueCount == 0");
+ }
+ }
+
+ protected void checkTypeName(PrimitiveType.PrimitiveTypeName expectedName) {
+ PrimitiveType.PrimitiveTypeName actualName = descriptor.getPrimitiveType().getPrimitiveTypeName();
+ Preconditions.checkArgument(
+ actualName == expectedName,
+ "Expected type name: %s, actual type name: %s",
+ expectedName,
+ actualName);
+ }
+
+ /**
+ * Reads `total` values from this columnReader into column.
+ */
+ @Override
+ public final void readToVector(int readNumber, V vector) throws IOException {
+ int rowId = 0;
+ WritableIntVector dictionaryIds = null;
+ if (dictionary != null) {
+ dictionaryIds = vector.reserveDictionaryIds(readNumber);
+ }
+ while (readNumber > 0) {
+ // Compute the number of values we want to read in this page.
+ int leftInPage = (int) (endOfPageValueCount - valuesRead);
+ if (leftInPage == 0) {
+ DataPage page = pageReader.readPage();
+ if (page instanceof DataPageV1) {
+ readPageV1((DataPageV1) page);
+ } else if (page instanceof DataPageV2) {
+ readPageV2((DataPageV2) page);
+ } else {
+ throw new RuntimeException("Unsupported page type: " + page.getClass());
+ }
+ leftInPage = (int) (endOfPageValueCount - valuesRead);
+ }
+ int num = Math.min(readNumber, leftInPage);
+ if (isCurrentPageDictionaryEncoded) {
+ // Read and decode dictionary ids.
+ runLenDecoder.readDictionaryIds(
+ num, dictionaryIds, vector, rowId, maxDefLevel, this.dictionaryIdsDecoder);
+
+ if (vector.hasDictionary() || (rowId == 0 && supportLazyDecode())) {
+ // Column vector supports lazy decoding of dictionary values so just set the dictionary.
+ // We can't do this if rowId != 0 AND the column doesn't have a dictionary (i.e. some
+ // non-dictionary encoded values have already been added).
+ vector.setDictionary(new ParquetDictionary(dictionary));
+ } else {
+ readBatchFromDictionaryIds(rowId, num, vector, dictionaryIds);
+ }
+ } else {
+ if (vector.hasDictionary() && rowId != 0) {
+ // This batch already has dictionary encoded values but this new page is not. The batch
+ // does not support a mix of dictionary and not so we will decode the dictionary.
+ readBatchFromDictionaryIds(0, rowId, vector, vector.getDictionaryIds());
+ }
+ vector.setDictionary(null);
+ readBatch(rowId, num, vector);
+ }
+
+ valuesRead += num;
+ rowId += num;
+ readNumber -= num;
+ }
+ }
+
+ private void readPageV1(DataPageV1 page) throws IOException {
+ this.pageValueCount = page.getValueCount();
+ ValuesReader rlReader = page.getRlEncoding().getValuesReader(descriptor, REPETITION_LEVEL);
+
+ // Initialize the decoders.
+ if (page.getDlEncoding() != Encoding.RLE && descriptor.getMaxDefinitionLevel() != 0) {
+ throw new UnsupportedOperationException("Unsupported encoding: " + page.getDlEncoding());
+ }
+ int bitWidth = BytesUtils.getWidthFromMaxInt(descriptor.getMaxDefinitionLevel());
+ this.runLenDecoder = new RunLengthDecoder(bitWidth);
+ try {
+ BytesInput bytes = page.getBytes();
+ ByteBufferInputStream in = bytes.toInputStream();
+ rlReader.initFromPage(pageValueCount, in);
+ this.runLenDecoder.initFromStream(pageValueCount, in);
+ prepareNewPage(page.getValueEncoding(), in);
+ } catch (IOException e) {
+ throw new IOException("could not read page " + page + " in col " + descriptor, e);
+ }
+ }
+
+ private void readPageV2(DataPageV2 page) throws IOException {
+ this.pageValueCount = page.getValueCount();
+
+ int bitWidth = BytesUtils.getWidthFromMaxInt(descriptor.getMaxDefinitionLevel());
+ // do not read the length from the stream. v2 pages handle dividing the page bytes.
+ this.runLenDecoder = new RunLengthDecoder(bitWidth, false);
+ this.runLenDecoder.initFromStream(
+ this.pageValueCount, page.getDefinitionLevels().toInputStream());
+ try {
+ prepareNewPage(page.getDataEncoding(), page.getData().toInputStream());
+ } catch (IOException e) {
+ throw new IOException("could not read page " + page + " in col " + descriptor, e);
+ }
+ }
+
+ private void prepareNewPage(
+ Encoding dataEncoding,
+ ByteBufferInputStream in) throws IOException {
+ this.endOfPageValueCount = valuesRead + pageValueCount;
+ if (dataEncoding.usesDictionary()) {
+ if (dictionary == null) {
+ throw new IOException("Could not read page in col "
+ + descriptor
+ + " as the dictionary was missing for encoding "
+ + dataEncoding);
+ }
+ @SuppressWarnings("deprecation")
+ Encoding plainDict = Encoding.PLAIN_DICTIONARY; // var to allow warning suppression
+ if (dataEncoding != plainDict && dataEncoding != Encoding.RLE_DICTIONARY) {
+ throw new UnsupportedOperationException("Unsupported encoding: " + dataEncoding);
+ }
+ this.dataInputStream = null;
+ this.dictionaryIdsDecoder = new RunLengthDecoder();
+ try {
+ this.dictionaryIdsDecoder.initFromStream(pageValueCount, in);
+ } catch (IOException e) {
+ throw new IOException("could not read dictionary in col " + descriptor, e);
+ }
+ this.isCurrentPageDictionaryEncoded = true;
+ } else {
+ if (dataEncoding != Encoding.PLAIN) {
+ throw new UnsupportedOperationException("Unsupported encoding: " + dataEncoding);
+ }
+ this.dictionaryIdsDecoder = null;
+ LOG.debug("init from page at offset {} for length {}", in.position(), in.available());
+ this.dataInputStream = in.remainingStream();
+ this.isCurrentPageDictionaryEncoded = false;
+ }
+
+ afterReadPage();
+ }
+
+ final ByteBuffer readDataBuffer(int length) {
+ try {
+ return dataInputStream.slice(length).order(ByteOrder.LITTLE_ENDIAN);
+ } catch (IOException e) {
+ throw new ParquetDecodingException("Failed to read " + length + " bytes", e);
+ }
+ }
+
+ /**
+ * After read a page, we may need some initialization.
+ */
+ protected void afterReadPage() {}
+
+ /**
+ * Support lazy dictionary ids decode. See more in {@link ParquetDictionary}.
+ * If return false, we will decode all the data first.
+ */
+ protected boolean supportLazyDecode() {
+ return true;
+ }
+
+ /**
+ * Read batch from {@link #runLenDecoder} and {@link #dataInputStream}.
+ */
+ protected abstract void readBatch(int rowId, int num, V column);
+
+ /**
+ * Decode dictionary ids to data.
+ * From {@link #runLenDecoder} and {@link #dictionaryIdsDecoder}.
+ */
+ protected abstract void readBatchFromDictionaryIds(
+ int rowId,
Review comment:
Can you refactor all the changes to follow the unified code style about the method parameters?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] danny0405 commented on a change in pull request #2613: [HUDI-1647] Supports snapshot read for Flink
Posted by GitBox <gi...@apache.org>.
danny0405 commented on a change in pull request #2613:
URL: https://github.com/apache/hudi/pull/2613#discussion_r587409966
##########
File path: hudi-flink/src/main/resources/META-INF/services/org.apache.flink.table.factories.TableFactory
##########
@@ -0,0 +1,17 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+org.apache.hudi.factory.HoodieTableFactory
Review comment:
Required for the java SPI service.
##########
File path: hudi-flink/src/test/resources/META-INF/services/org.apache.flink.table.factories.TableFactory
##########
@@ -0,0 +1,18 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+org.apache.hudi.factory.HoodieTableFactory
+org.apache.hudi.utils.factory.ContinuousFileSourceFactory
Review comment:
Required for the java SPI service.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] danny0405 commented on a change in pull request #2613: [HUDI-1647] Supports snapshot read for Flink
Posted by GitBox <gi...@apache.org>.
danny0405 commented on a change in pull request #2613:
URL: https://github.com/apache/hudi/pull/2613#discussion_r585212322
##########
File path: hudi-flink/src/main/java/org/apache/hudi/source/HoodieTableSource.java
##########
@@ -0,0 +1,411 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.source;
+
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.model.HoodieBaseFile;
+import org.apache.hudi.common.model.HoodieLogFile;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.TableSchemaResolver;
+import org.apache.hudi.common.table.view.HoodieTableFileSystemView;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.hadoop.HoodieROTablePathFilter;
+import org.apache.hudi.hadoop.utils.HoodieRealtimeInputFormatUtils;
+import org.apache.hudi.operator.FlinkOptions;
+import org.apache.hudi.source.format.FilePathUtils;
+import org.apache.hudi.source.format.cow.CopyOnWriteInputFormat;
+import org.apache.hudi.source.format.mor.MergeOnReadInputFormat;
+import org.apache.hudi.source.format.mor.MergeOnReadInputSplit;
+import org.apache.hudi.source.format.mor.MergeOnReadTableState;
+import org.apache.hudi.util.AvroSchemaConverter;
+import org.apache.hudi.util.StreamerUtil;
+
+import org.apache.avro.Schema;
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.common.io.FileInputFormat;
+import org.apache.flink.api.common.io.FilePathFilter;
+import org.apache.flink.api.common.io.InputFormat;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.io.CollectionInputFormat;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.DataStreamSource;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.source.InputFormatSourceFunction;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.expressions.Expression;
+import org.apache.flink.table.runtime.types.TypeInfoDataTypeConverter;
+import org.apache.flink.table.sources.FilterableTableSource;
+import org.apache.flink.table.sources.LimitableTableSource;
+import org.apache.flink.table.sources.PartitionableTableSource;
+import org.apache.flink.table.sources.ProjectableTableSource;
+import org.apache.flink.table.sources.StreamTableSource;
+import org.apache.flink.table.sources.TableSource;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.mapred.JobConf;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import static org.apache.hudi.hadoop.utils.HoodieRealtimeRecordReaderUtils.getMaxCompactionMemoryInBytes;
+import static org.apache.hudi.source.format.FormatUtils.getParquetConf;
+
+/**
+ * Hoodie batch table source that always read the latest snapshot of the underneath table.
+ */
+public class HoodieTableSource implements
+ StreamTableSource<RowData>,
+ PartitionableTableSource,
+ ProjectableTableSource<RowData>,
+ LimitableTableSource<RowData>,
+ FilterableTableSource<RowData> {
+ private static final Logger LOG = LoggerFactory.getLogger(HoodieTableSource.class);
+
+ private final transient org.apache.hadoop.conf.Configuration hadoopConf;
+ private final transient HoodieTableMetaClient metaClient;
+ private final long maxCompactionMemoryInBytes;
+
+ private final TableSchema schema;
+ private final Path path;
+ private final List<String> partitionKeys;
+ private final String defaultPartName;
+ private final Configuration conf;
+
+ private final int[] requiredPos;
+ private final long limit;
+ private final List<Expression> filters;
+
+ private List<Map<String, String>> requiredPartitions;
+
+ public HoodieTableSource(
+ TableSchema schema,
+ Path path,
+ List<String> partitionKeys,
+ String defaultPartName,
+ Configuration conf) {
+ this(schema, path, partitionKeys, defaultPartName, conf,
+ null, null, null, null);
+ }
+
+ public HoodieTableSource(
+ TableSchema schema,
+ Path path,
+ List<String> partitionKeys,
+ String defaultPartName,
+ Configuration conf,
+ @Nullable List<Map<String, String>> requiredPartitions,
+ @Nullable int[] requiredPos,
+ @Nullable Long limit,
+ @Nullable List<Expression> filters) {
+ this.schema = schema;
+ this.path = path;
+ this.partitionKeys = partitionKeys;
+ this.defaultPartName = defaultPartName;
+ this.conf = conf;
+ this.requiredPartitions = requiredPartitions;
+ this.requiredPos = requiredPos == null
+ ? IntStream.range(0, schema.getFieldCount()).toArray()
+ : requiredPos;
+ this.limit = limit == null ? -1 : limit;
+ this.filters = filters == null ? Collections.emptyList() : filters;
+ final String basePath = this.conf.getString(FlinkOptions.PATH);
+ this.hadoopConf = StreamerUtil.getHadoopConf();
+ this.metaClient = HoodieTableMetaClient.builder().setConf(hadoopConf).setBasePath(basePath).build();
+ this.maxCompactionMemoryInBytes = getMaxCompactionMemoryInBytes(new JobConf(this.hadoopConf));
+ }
+
+ @Override
+ public DataStream<RowData> getDataStream(StreamExecutionEnvironment execEnv) {
+ @SuppressWarnings("unchecked")
+ TypeInformation<RowData> typeInfo =
+ (TypeInformation<RowData>) TypeInfoDataTypeConverter.fromDataTypeToTypeInfo(getProducedDataType());
+ InputFormatSourceFunction<RowData> func = new InputFormatSourceFunction<>(getInputFormat(), typeInfo);
+ DataStreamSource<RowData> source = execEnv.addSource(func, explainSource(), typeInfo);
+ return source.name(explainSource());
+ }
+
+ @Override
+ public boolean isBounded() {
+ return true;
+ }
+
+ @Override
+ public TableSource<RowData> applyPredicate(List<Expression> predicates) {
+ return new HoodieTableSource(schema, path, partitionKeys, defaultPartName, conf,
+ requiredPartitions, requiredPos, limit, new ArrayList<>(predicates));
+ }
+
+ @Override
+ public boolean isFilterPushedDown() {
+ return this.filters != null && this.filters.size() > 0;
+ }
+
+ @Override
+ public boolean isLimitPushedDown() {
+ return this.limit != -1;
+ }
+
+ @Override
+ public TableSource<RowData> applyLimit(long limit) {
+ return new HoodieTableSource(schema, path, partitionKeys, defaultPartName, conf,
+ requiredPartitions, requiredPos, limit, filters);
+ }
+
+ @Override
+ public List<Map<String, String>> getPartitions() {
+ try {
+ return FilePathUtils
+ .searchPartKeyValueAndPaths(
+ path.getFileSystem(),
+ path,
+ conf.getBoolean(FlinkOptions.HIVE_STYLE_PARTITION),
+ partitionKeys.toArray(new String[0]))
+ .stream()
+ .map(tuple2 -> tuple2.f0)
+ .map(spec -> {
+ LinkedHashMap<String, String> ret = new LinkedHashMap<>();
+ spec.forEach((k, v) -> ret.put(k, defaultPartName.equals(v) ? null : v));
+ return ret;
+ })
+ .collect(Collectors.toList());
+ } catch (Exception e) {
+ throw new TableException("Fetch partitions fail.", e);
+ }
+ }
+
+ @Override
+ public TableSource applyPartitionPruning(List<Map<String, String>> requiredPartitions) {
+ return new HoodieTableSource(schema, path, partitionKeys, defaultPartName, conf,
+ requiredPartitions, requiredPos, limit, filters);
+ }
+
+ @Override
+ public TableSource<RowData> projectFields(int[] requiredPos) {
+ return new HoodieTableSource(schema, path, partitionKeys, defaultPartName, conf,
+ requiredPartitions, requiredPos, limit, filters);
+ }
+
+ @Override
+ public TableSchema getTableSchema() {
+ return schema;
+ }
+
+ @Override
+ public DataType getProducedDataType() {
+ String[] schemaFieldNames = this.schema.getFieldNames();
+ DataType[] schemaTypes = this.schema.getFieldDataTypes();
+
+ return DataTypes.ROW(Arrays.stream(this.requiredPos)
+ .mapToObj(i -> DataTypes.FIELD(schemaFieldNames[i], schemaTypes[i]))
+ .toArray(DataTypes.Field[]::new))
+ .bridgedTo(RowData.class);
+ }
+
+ private List<Map<String, String>> getOrFetchPartitions() {
+ if (requiredPartitions == null) {
+ requiredPartitions = getPartitions();
+ }
+ return requiredPartitions;
+ }
+
+ private List<MergeOnReadInputSplit> buildFileIndex(Path[] paths) {
+ FileStatus[] fileStatuses = Arrays.stream(paths)
+ .flatMap(path -> Arrays.stream(FilePathUtils.getFileStatusRecursivelyV2(path, 1, hadoopConf)))
+ .toArray(FileStatus[]::new);
+ if (fileStatuses.length == 0) {
+ throw new HoodieException("No files found for reading in user provided path.");
+ }
+
+ HoodieTableFileSystemView fsView = new HoodieTableFileSystemView(metaClient,
+ metaClient.getActiveTimeline().getCommitsTimeline()
+ .filterCompletedInstants(), fileStatuses);
+ List<HoodieBaseFile> latestFiles = fsView.getLatestBaseFiles().collect(Collectors.toList());
+ String latestCommit = fsView.getLastInstant().get().getTimestamp();
+ final String mergeType = this.conf.getString(FlinkOptions.MERGE_TYPE);
+ final AtomicInteger cnt = new AtomicInteger(0);
+ if (latestFiles.size() > 0) {
+ Map<HoodieBaseFile, List<String>> fileGroup =
+ HoodieRealtimeInputFormatUtils.groupLogsByBaseFile(hadoopConf, latestFiles);
+ return fileGroup.entrySet().stream().map(kv -> {
+ HoodieBaseFile baseFile = kv.getKey();
+ Option<List<String>> logPaths = kv.getValue().size() == 0
+ ? Option.empty()
+ : Option.of(kv.getValue());
+ return new MergeOnReadInputSplit(cnt.getAndAdd(1),
+ baseFile.getPath(), logPaths, latestCommit,
+ metaClient.getBasePath(), maxCompactionMemoryInBytes, mergeType);
+ }).collect(Collectors.toList());
+ } else {
+ // all the files are logs
+ return Arrays.stream(paths).map(partitionPath -> {
+ String relPartitionPath = FSUtils.getRelativePartitionPath(
+ new org.apache.hadoop.fs.Path(path.toUri()),
+ new org.apache.hadoop.fs.Path(partitionPath.toUri()));
+ return fsView.getLatestMergedFileSlicesBeforeOrOn(relPartitionPath, latestCommit)
+ .map(fileSlice -> {
+ Option<List<String>> logPaths = Option.ofNullable(fileSlice.getLogFiles()
+ .sorted(HoodieLogFile.getLogFileComparator())
+ .map(logFile -> logFile.getPath().toString())
+ .collect(Collectors.toList()));
+ return new MergeOnReadInputSplit(cnt.getAndAdd(1),
+ null, logPaths, latestCommit,
+ metaClient.getBasePath(), maxCompactionMemoryInBytes, mergeType);
+ }).collect(Collectors.toList()); })
+ .flatMap(Collection::stream)
+ .collect(Collectors.toList());
+ }
+ }
+
+ @VisibleForTesting
+ public InputFormat<RowData, ?> getInputFormat() {
+ // When this table has no partition, just return an empty source.
+ if (!partitionKeys.isEmpty() && getOrFetchPartitions().isEmpty()) {
+ return new CollectionInputFormat<>(Collections.emptyList(), null);
+ }
+
+ final Path[] paths = getReadPaths();
+ if (paths.length == 0) {
+ return new CollectionInputFormat<>(Collections.emptyList(), null);
+ }
+
+ TableSchemaResolver schemaUtil = new TableSchemaResolver(metaClient);
+ final Schema tableAvroSchema;
+ try {
+ tableAvroSchema = schemaUtil.getTableAvroSchema();
+ } catch (Exception e) {
+ throw new HoodieException("Get table avro schema error", e);
+ }
+ final DataType rowDataType = AvroSchemaConverter.convertToDataType(tableAvroSchema);
+ final RowType rowType = (RowType) rowDataType.getLogicalType();
+ final RowType requiredRowType = (RowType) getProducedDataType().notNull().getLogicalType();
+
+ final String queryType = this.conf.getString(FlinkOptions.QUERY_TYPE);
+ if (queryType.equals(FlinkOptions.QUERY_TYPE_SNAPSHOT)) {
+ switch (this.conf.getString(FlinkOptions.TABLE_TYPE)) {
+ case FlinkOptions.TABLE_TYPE_MERGE_ON_READ:
Review comment:
I found that a constant in `FlinkOptions` is more friendly to use.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] danny0405 commented on a change in pull request #2613: [HUDI-1647] Supports snapshot read for Flink
Posted by GitBox <gi...@apache.org>.
danny0405 commented on a change in pull request #2613:
URL: https://github.com/apache/hudi/pull/2613#discussion_r586099830
##########
File path: hudi-flink/src/main/java/org/apache/hudi/source/format/FormatUtils.java
##########
@@ -0,0 +1,98 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.source.format;
+
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.table.log.HoodieMergedLogRecordScanner;
+import org.apache.hudi.hadoop.config.HoodieRealtimeConfig;
+import org.apache.hudi.operator.FlinkOptions;
+import org.apache.hudi.source.format.mor.MergeOnReadInputSplit;
+
+import org.apache.avro.Schema;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.avro.generic.GenericRecordBuilder;
+import org.apache.avro.generic.IndexedRecord;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+
+import java.util.Arrays;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+
+/**
+ * Utilities for format.
+ */
+public class FormatUtils {
+ private FormatUtils() {
+ }
+
+ public static GenericRecord buildAvroRecordBySchema(
+ IndexedRecord record,
+ Schema requiredSchema,
+ int[] requiredPos,
+ GenericRecordBuilder recordBuilder) {
+ List<Schema.Field> requiredFields = requiredSchema.getFields();
+ assert (requiredFields.size() == requiredPos.length);
+ Iterator<Integer> positionIterator = Arrays.stream(requiredPos).iterator();
+ requiredFields.forEach(f -> recordBuilder.set(f, record.get(positionIterator.next())));
+ return recordBuilder.build();
+ }
+
+ public static HoodieMergedLogRecordScanner scanLog(
+ MergeOnReadInputSplit split,
+ Schema logSchema,
+ Configuration config) {
+ FileSystem fs = FSUtils.getFs(split.getTablePath(), config);
+ return HoodieMergedLogRecordScanner.newBuilder()
Review comment:
Yes, we can promote it in the future, i'm not planning to do it in this PR.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] danny0405 commented on a change in pull request #2613: [HUDI-1647] Supports snapshot read for Flink
Posted by GitBox <gi...@apache.org>.
danny0405 commented on a change in pull request #2613:
URL: https://github.com/apache/hudi/pull/2613#discussion_r586099102
##########
File path: hudi-flink/src/main/java/org/apache/hudi/sink/HoodieTableSink.java
##########
@@ -0,0 +1,126 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.sink;
+
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.operator.FlinkOptions;
+import org.apache.hudi.operator.StreamWriteOperatorFactory;
+import org.apache.hudi.operator.compact.CompactFunction;
+import org.apache.hudi.operator.compact.CompactionCommitEvent;
+import org.apache.hudi.operator.compact.CompactionCommitSink;
+import org.apache.hudi.operator.compact.CompactionPlanEvent;
+import org.apache.hudi.operator.compact.CompactionPlanOperator;
+import org.apache.hudi.operator.partitioner.BucketAssignFunction;
+import org.apache.hudi.operator.transform.RowDataToHoodieFunction;
+import org.apache.hudi.util.StreamerUtil;
+
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.DataStreamSink;
+import org.apache.flink.streaming.api.functions.sink.SinkFunction;
+import org.apache.flink.streaming.api.operators.KeyedProcessOperator;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.sinks.AppendStreamTableSink;
+import org.apache.flink.table.sinks.PartitionableTableSink;
+import org.apache.flink.table.sinks.TableSink;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.RowType;
+
+import java.util.Map;
+
+/**
+ * Hoodie table sink.
+ */
+public class HoodieTableSink implements AppendStreamTableSink<RowData>, PartitionableTableSink {
Review comment:
For both batch and streaming.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] codecov-io commented on pull request #2613: [HUDI-1647] Supports snapshot read for Flink
Posted by GitBox <gi...@apache.org>.
codecov-io commented on pull request #2613:
URL: https://github.com/apache/hudi/pull/2613#issuecomment-787825622
# [Codecov](https://codecov.io/gh/apache/hudi/pull/2613?src=pr&el=h1) Report
> Merging [#2613](https://codecov.io/gh/apache/hudi/pull/2613?src=pr&el=desc) (99a6c1f) into [master](https://codecov.io/gh/apache/hudi/commit/7a11de12764d8f68f296c6e68a22822318bfbefa?el=desc) (7a11de1) will **increase** coverage by `17.97%`.
> The diff coverage is `n/a`.
[](https://codecov.io/gh/apache/hudi/pull/2613?src=pr&el=tree)
```diff
@@ Coverage Diff @@
## master #2613 +/- ##
=============================================
+ Coverage 51.56% 69.54% +17.97%
+ Complexity 3286 363 -2923
=============================================
Files 445 53 -392
Lines 20328 1944 -18384
Branches 2102 235 -1867
=============================================
- Hits 10483 1352 -9131
+ Misses 8978 458 -8520
+ Partials 867 134 -733
```
| Flag | Coverage Δ | Complexity Δ | |
|---|---|---|---|
| hudicli | `?` | `?` | |
| hudiclient | `?` | `?` | |
| hudicommon | `?` | `?` | |
| hudiflink | `?` | `?` | |
| hudihadoopmr | `?` | `?` | |
| hudisparkdatasource | `?` | `?` | |
| hudisync | `?` | `?` | |
| huditimelineservice | `?` | `?` | |
| hudiutilities | `69.54% <ø> (ø)` | `0.00 <ø> (ø)` | |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags#carryforward-flags-in-the-pull-request-comment) to find out more.
| [Impacted Files](https://codecov.io/gh/apache/hudi/pull/2613?src=pr&el=tree) | Coverage Δ | Complexity Δ | |
|---|---|---|---|
| [...rc/main/scala/org/apache/hudi/Spark3RowSerDe.scala](https://codecov.io/gh/apache/hudi/pull/2613/diff?src=pr&el=tree#diff-aHVkaS1zcGFyay1kYXRhc291cmNlL2h1ZGktc3BhcmszL3NyYy9tYWluL3NjYWxhL29yZy9hcGFjaGUvaHVkaS9TcGFyazNSb3dTZXJEZS5zY2FsYQ==) | | | |
| [...hudi/common/model/HoodieReplaceCommitMetadata.java](https://codecov.io/gh/apache/hudi/pull/2613/diff?src=pr&el=tree#diff-aHVkaS1jb21tb24vc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2h1ZGkvY29tbW9uL21vZGVsL0hvb2RpZVJlcGxhY2VDb21taXRNZXRhZGF0YS5qYXZh) | | | |
| [...in/java/org/apache/hudi/cli/HoodiePrintHelper.java](https://codecov.io/gh/apache/hudi/pull/2613/diff?src=pr&el=tree#diff-aHVkaS1jbGkvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2h1ZGkvY2xpL0hvb2RpZVByaW50SGVscGVyLmphdmE=) | | | |
| [...pache/hudi/operator/KeyedWriteProcessOperator.java](https://codecov.io/gh/apache/hudi/pull/2613/diff?src=pr&el=tree#diff-aHVkaS1mbGluay9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvaHVkaS9vcGVyYXRvci9LZXllZFdyaXRlUHJvY2Vzc09wZXJhdG9yLmphdmE=) | | | |
| [.../hudi/common/bloom/InternalDynamicBloomFilter.java](https://codecov.io/gh/apache/hudi/pull/2613/diff?src=pr&el=tree#diff-aHVkaS1jb21tb24vc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2h1ZGkvY29tbW9uL2Jsb29tL0ludGVybmFsRHluYW1pY0Jsb29tRmlsdGVyLmphdmE=) | | | |
| [...util/jvm/OpenJ9MemoryLayoutSpecification32bit.java](https://codecov.io/gh/apache/hudi/pull/2613/diff?src=pr&el=tree#diff-aHVkaS1jb21tb24vc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2h1ZGkvY29tbW9uL3V0aWwvanZtL09wZW5KOU1lbW9yeUxheW91dFNwZWNpZmljYXRpb24zMmJpdC5qYXZh) | | | |
| [.../hadoop/realtime/RealtimeUnmergedRecordReader.java](https://codecov.io/gh/apache/hudi/pull/2613/diff?src=pr&el=tree#diff-aHVkaS1oYWRvb3AtbXIvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2h1ZGkvaGFkb29wL3JlYWx0aW1lL1JlYWx0aW1lVW5tZXJnZWRSZWNvcmRSZWFkZXIuamF2YQ==) | | | |
| [...del/OverwriteNonDefaultsWithLatestAvroPayload.java](https://codecov.io/gh/apache/hudi/pull/2613/diff?src=pr&el=tree#diff-aHVkaS1jb21tb24vc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2h1ZGkvY29tbW9uL21vZGVsL092ZXJ3cml0ZU5vbkRlZmF1bHRzV2l0aExhdGVzdEF2cm9QYXlsb2FkLmphdmE=) | | | |
| [...va/org/apache/hudi/cli/commands/CleansCommand.java](https://codecov.io/gh/apache/hudi/pull/2613/diff?src=pr&el=tree#diff-aHVkaS1jbGkvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2h1ZGkvY2xpL2NvbW1hbmRzL0NsZWFuc0NvbW1hbmQuamF2YQ==) | | | |
| [...che/hudi/common/table/timeline/dto/LogFileDTO.java](https://codecov.io/gh/apache/hudi/pull/2613/diff?src=pr&el=tree#diff-aHVkaS1jb21tb24vc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2h1ZGkvY29tbW9uL3RhYmxlL3RpbWVsaW5lL2R0by9Mb2dGaWxlRFRPLmphdmE=) | | | |
| ... and [374 more](https://codecov.io/gh/apache/hudi/pull/2613/diff?src=pr&el=tree-more) | |
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] danny0405 commented on a change in pull request #2613: [HUDI-1647] Supports snapshot read for Flink
Posted by GitBox <gi...@apache.org>.
danny0405 commented on a change in pull request #2613:
URL: https://github.com/apache/hudi/pull/2613#discussion_r586100912
##########
File path: hudi-flink/src/main/java/org/apache/hudi/source/format/cow/ParquetColumnarRowSplitReader.java
##########
@@ -0,0 +1,370 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.source.format.cow;
+
+import org.apache.flink.formats.parquet.vector.reader.ColumnReader;
+import org.apache.flink.table.data.ColumnarRowData;
+import org.apache.flink.table.data.vector.ColumnVector;
+import org.apache.flink.table.data.vector.VectorizedColumnBatch;
+import org.apache.flink.table.data.vector.writable.WritableColumnVector;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
+import org.apache.flink.util.FlinkRuntimeException;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.parquet.column.ColumnDescriptor;
+import org.apache.parquet.column.page.PageReadStore;
+import org.apache.parquet.filter2.compat.FilterCompat;
+import org.apache.parquet.hadoop.ParquetFileReader;
+import org.apache.parquet.hadoop.metadata.BlockMetaData;
+import org.apache.parquet.hadoop.metadata.ParquetMetadata;
+import org.apache.parquet.schema.GroupType;
+import org.apache.parquet.schema.MessageType;
+import org.apache.parquet.schema.Type;
+import org.apache.parquet.schema.Types;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+
+import static org.apache.hudi.source.format.cow.ParquetSplitReaderUtil.createColumnReader;
+import static org.apache.hudi.source.format.cow.ParquetSplitReaderUtil.createWritableColumnVector;
+import static org.apache.parquet.filter2.compat.RowGroupFilter.filterRowGroups;
+import static org.apache.parquet.format.converter.ParquetMetadataConverter.range;
+import static org.apache.parquet.hadoop.ParquetFileReader.readFooter;
+import static org.apache.parquet.hadoop.ParquetInputFormat.getFilter;
+
+/**
+ * This reader is used to read a {@link VectorizedColumnBatch} from input split.
+ *
+ * <p>Note: Reference Flink release 1.11.2
+ * {@code org.apache.flink.formats.parquet.vector.ParquetColumnarRowSplitReader}
+ * because it is package scope.
+ */
+public class ParquetColumnarRowSplitReader implements Closeable {
Review comment:
It is not that good to put the copied code, but it is still better than adding `flink-parquet` dependency and shade all the codes in the class path, which would cause conflicts and confuses the users (you can not prevent people to have the original `flink-parquet` jar in the classpath).
By copy several classes we avoid that confusion.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] garyli1019 commented on a change in pull request #2613: [HUDI-1647] Supports snapshot read for Flink
Posted by GitBox <gi...@apache.org>.
garyli1019 commented on a change in pull request #2613:
URL: https://github.com/apache/hudi/pull/2613#discussion_r587262267
##########
File path: hudi-flink/src/main/resources/META-INF/services/org.apache.flink.table.factories.TableFactory
##########
@@ -0,0 +1,17 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+org.apache.hudi.factory.HoodieTableFactory
Review comment:
is this file necessary?
##########
File path: hudi-flink/src/test/java/org/apache/hudi/source/HoodieDataSourceITCase.java
##########
@@ -0,0 +1,162 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.source;
+
+import org.apache.hudi.operator.FlinkOptions;
+import org.apache.hudi.operator.utils.TestConfigurations;
+
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.TableResult;
+import org.apache.flink.table.api.config.ExecutionConfigOptions;
+import org.apache.flink.table.api.internal.TableEnvironmentImpl;
+import org.apache.flink.test.util.AbstractTestBase;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.CollectionUtil;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.File;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.ExecutionException;
+import java.util.stream.Collectors;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+
+/**
+ * IT cases for Hoodie table source and sink.
+ *
+ * Note: should add more SQL cases when batch write is supported.
+ */
+public class HoodieDataSourceITCase extends AbstractTestBase {
+ private TableEnvironment streamTableEnv;
+ private TableEnvironment batchTableEnv;
+
+ @BeforeEach
+ void beforeEach() {
+ EnvironmentSettings settings = EnvironmentSettings.newInstance().build();
+ streamTableEnv = TableEnvironmentImpl.create(settings);
+ streamTableEnv.getConfig().getConfiguration()
+ .setInteger(ExecutionConfigOptions.TABLE_EXEC_RESOURCE_DEFAULT_PARALLELISM, 1);
+ streamTableEnv.getConfig().getConfiguration()
+ .setString("execution.checkpointing.interval", "2s");
+
+ settings = EnvironmentSettings.newInstance().inBatchMode().build();
+ batchTableEnv = TableEnvironmentImpl.create(settings);
+ batchTableEnv.getConfig().getConfiguration()
+ .setInteger(ExecutionConfigOptions.TABLE_EXEC_RESOURCE_DEFAULT_PARALLELISM, 1);
+ }
+
+ @TempDir
+ File tempFile;
+
+ @Test
+ void testStreamWriteBatchRead() {
Review comment:
is that possible to add more test cases to cover COW/MOR(without comapction)/MOR(with compaction) and query?
##########
File path: hudi-flink/src/test/resources/META-INF/services/org.apache.flink.table.factories.TableFactory
##########
@@ -0,0 +1,18 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+org.apache.hudi.factory.HoodieTableFactory
+org.apache.hudi.utils.factory.ContinuousFileSourceFactory
Review comment:
ditto
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] danny0405 commented on a change in pull request #2613: [HUDI-1647] Supports snapshot read for Flink
Posted by GitBox <gi...@apache.org>.
danny0405 commented on a change in pull request #2613:
URL: https://github.com/apache/hudi/pull/2613#discussion_r586099830
##########
File path: hudi-flink/src/main/java/org/apache/hudi/source/format/FormatUtils.java
##########
@@ -0,0 +1,98 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.source.format;
+
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.table.log.HoodieMergedLogRecordScanner;
+import org.apache.hudi.hadoop.config.HoodieRealtimeConfig;
+import org.apache.hudi.operator.FlinkOptions;
+import org.apache.hudi.source.format.mor.MergeOnReadInputSplit;
+
+import org.apache.avro.Schema;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.avro.generic.GenericRecordBuilder;
+import org.apache.avro.generic.IndexedRecord;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+
+import java.util.Arrays;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+
+/**
+ * Utilities for format.
+ */
+public class FormatUtils {
+ private FormatUtils() {
+ }
+
+ public static GenericRecord buildAvroRecordBySchema(
+ IndexedRecord record,
+ Schema requiredSchema,
+ int[] requiredPos,
+ GenericRecordBuilder recordBuilder) {
+ List<Schema.Field> requiredFields = requiredSchema.getFields();
+ assert (requiredFields.size() == requiredPos.length);
+ Iterator<Integer> positionIterator = Arrays.stream(requiredPos).iterator();
+ requiredFields.forEach(f -> recordBuilder.set(f, record.get(positionIterator.next())));
+ return recordBuilder.build();
+ }
+
+ public static HoodieMergedLogRecordScanner scanLog(
+ MergeOnReadInputSplit split,
+ Schema logSchema,
+ Configuration config) {
+ FileSystem fs = FSUtils.getFs(split.getTablePath(), config);
+ return HoodieMergedLogRecordScanner.newBuilder()
Review comment:
Yes, we can promote it in the future, i'm not plan to do it in this PR.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] danny0405 commented on a change in pull request #2613: [HUDI-1647] Supports snapshot read for Flink
Posted by GitBox <gi...@apache.org>.
danny0405 commented on a change in pull request #2613:
URL: https://github.com/apache/hudi/pull/2613#discussion_r586098951
##########
File path: hudi-flink/src/main/java/org/apache/hudi/source/HoodieTableSource.java
##########
@@ -0,0 +1,410 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.source;
+
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.model.HoodieBaseFile;
+import org.apache.hudi.common.model.HoodieLogFile;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.TableSchemaResolver;
+import org.apache.hudi.common.table.view.HoodieTableFileSystemView;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.hadoop.HoodieROTablePathFilter;
+import org.apache.hudi.hadoop.utils.HoodieRealtimeInputFormatUtils;
+import org.apache.hudi.operator.FlinkOptions;
+import org.apache.hudi.source.format.FilePathUtils;
+import org.apache.hudi.source.format.cow.CopyOnWriteInputFormat;
+import org.apache.hudi.source.format.mor.MergeOnReadInputFormat;
+import org.apache.hudi.source.format.mor.MergeOnReadInputSplit;
+import org.apache.hudi.source.format.mor.MergeOnReadTableState;
+import org.apache.hudi.util.AvroSchemaConverter;
+import org.apache.hudi.util.StreamerUtil;
+
+import org.apache.avro.Schema;
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.common.io.FileInputFormat;
+import org.apache.flink.api.common.io.FilePathFilter;
+import org.apache.flink.api.common.io.InputFormat;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.io.CollectionInputFormat;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.DataStreamSource;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.source.InputFormatSourceFunction;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.expressions.Expression;
+import org.apache.flink.table.runtime.types.TypeInfoDataTypeConverter;
+import org.apache.flink.table.sources.FilterableTableSource;
+import org.apache.flink.table.sources.LimitableTableSource;
+import org.apache.flink.table.sources.PartitionableTableSource;
+import org.apache.flink.table.sources.ProjectableTableSource;
+import org.apache.flink.table.sources.StreamTableSource;
+import org.apache.flink.table.sources.TableSource;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.mapred.JobConf;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import static org.apache.hudi.hadoop.utils.HoodieRealtimeRecordReaderUtils.getMaxCompactionMemoryInBytes;
+import static org.apache.hudi.source.format.FormatUtils.getParquetConf;
+
+/**
+ * Hoodie batch table source that always read the latest snapshot of the underneath table.
+ */
+public class HoodieTableSource implements
+ StreamTableSource<RowData>,
+ PartitionableTableSource,
+ ProjectableTableSource<RowData>,
+ LimitableTableSource<RowData>,
+ FilterableTableSource<RowData> {
+ private static final Logger LOG = LoggerFactory.getLogger(HoodieTableSource.class);
+
+ private final transient org.apache.hadoop.conf.Configuration hadoopConf;
+ private final transient HoodieTableMetaClient metaClient;
+ private final long maxCompactionMemoryInBytes;
+
+ private final TableSchema schema;
+ private final Path path;
+ private final List<String> partitionKeys;
+ private final String defaultPartName;
+ private final Configuration conf;
+
+ private final int[] requiredPos;
+ private final long limit;
+ private final List<Expression> filters;
+
+ private List<Map<String, String>> requiredPartitions;
+
+ public HoodieTableSource(
+ TableSchema schema,
Review comment:
HUDI flink also uses `TableSchemaResolver`, see `HoodieTableSource.getInputFormat`.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] yanghua commented on a change in pull request #2613: [HUDI-1647] Supports snapshot read for Flink
Posted by GitBox <gi...@apache.org>.
yanghua commented on a change in pull request #2613:
URL: https://github.com/apache/hudi/pull/2613#discussion_r584548701
##########
File path: hudi-flink/src/main/java/org/apache/hudi/operator/FlinkOptions.java
##########
@@ -64,6 +74,41 @@ private FlinkOptions() {
.noDefaultValue()
.withDescription("Avro schema file path, the parsed schema is used for deserializing");
+ public static final String QUERY_TYPE_SNAPSHOT = "snapshot";
+ public static final String QUERY_TYPE_READ_OPTIMIZED = "read_optimized";
+ public static final String QUERY_TYPE_INCREMENTAL = "incremental";
+ public static final ConfigOption<String> QUERY_TYPE = ConfigOptions
+ .key("hoodie.datasource.query.type")
+ .stringType()
+ .defaultValue(QUERY_TYPE_SNAPSHOT)
+ .withDescription("Decides how data files need to be read, in\n"
+ + "1) Snapshot mode (obtain latest view, based on row & columnar data);\n"
+ + "2) incremental mode (new data since an instantTime);\n"
+ + "3) Read Optimized mode (obtain latest view, based on columnar data)\n."
+ + "Default: snapshot");
+
+ public static final String REALTIME_SKIP_MERGE = "skip_merge";
+ public static final String REALTIME_PAYLOAD_COMBINE = "payload_combine";
+ public static final ConfigOption<String> MERGE_TYPE = ConfigOptions
+ .key("hoodie.datasource.merge.type")
+ .stringType()
+ .defaultValue(REALTIME_PAYLOAD_COMBINE)
+ .withDescription("For Snapshot query on merge on read table. Use this key to define the payload class");
Review comment:
`define the payload class` seems to make the semantic not clear. What about replacing with `define the merge type`?
##########
File path: hudi-flink/src/main/java/org/apache/hudi/factory/HoodieTableFactory.java
##########
@@ -0,0 +1,84 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.factory;
+
+import org.apache.hudi.operator.FlinkOptions;
+import org.apache.hudi.sink.HoodieTableSink;
+import org.apache.hudi.source.HoodieTableSource;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.factories.FactoryUtil;
+import org.apache.flink.table.factories.TableSinkFactory;
+import org.apache.flink.table.factories.TableSourceFactory;
+import org.apache.flink.table.sinks.TableSink;
+import org.apache.flink.table.sources.TableSource;
+import org.apache.flink.table.utils.TableSchemaUtils;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Hoodie data source table factory.
Review comment:
This class provided the function for creating both source and sink. Shall we change this comment to `Hoodie table factory`?
##########
File path: hudi-flink/src/main/java/org/apache/hudi/operator/FlinkOptions.java
##########
@@ -64,6 +74,41 @@ private FlinkOptions() {
.noDefaultValue()
.withDescription("Avro schema file path, the parsed schema is used for deserializing");
+ public static final String QUERY_TYPE_SNAPSHOT = "snapshot";
+ public static final String QUERY_TYPE_READ_OPTIMIZED = "read_optimized";
+ public static final String QUERY_TYPE_INCREMENTAL = "incremental";
+ public static final ConfigOption<String> QUERY_TYPE = ConfigOptions
+ .key("hoodie.datasource.query.type")
+ .stringType()
+ .defaultValue(QUERY_TYPE_SNAPSHOT)
+ .withDescription("Decides how data files need to be read, in\n"
+ + "1) Snapshot mode (obtain latest view, based on row & columnar data);\n"
+ + "2) incremental mode (new data since an instantTime);\n"
+ + "3) Read Optimized mode (obtain latest view, based on columnar data)\n."
+ + "Default: snapshot");
+
+ public static final String REALTIME_SKIP_MERGE = "skip_merge";
+ public static final String REALTIME_PAYLOAD_COMBINE = "payload_combine";
+ public static final ConfigOption<String> MERGE_TYPE = ConfigOptions
+ .key("hoodie.datasource.merge.type")
+ .stringType()
+ .defaultValue(REALTIME_PAYLOAD_COMBINE)
+ .withDescription("For Snapshot query on merge on read table. Use this key to define the payload class");
+
+ public static final ConfigOption<Boolean> HIVE_STYLE_PARTITION = ConfigOptions
+ .key("hoodie.datasource.hive-style.partition")
Review comment:
Can we ignore `-` in the config key?
##########
File path: hudi-flink/src/main/java/org/apache/hudi/source/HoodieTableSource.java
##########
@@ -0,0 +1,411 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.source;
+
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.model.HoodieBaseFile;
+import org.apache.hudi.common.model.HoodieLogFile;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.TableSchemaResolver;
+import org.apache.hudi.common.table.view.HoodieTableFileSystemView;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.hadoop.HoodieROTablePathFilter;
+import org.apache.hudi.hadoop.utils.HoodieRealtimeInputFormatUtils;
+import org.apache.hudi.operator.FlinkOptions;
+import org.apache.hudi.source.format.FilePathUtils;
+import org.apache.hudi.source.format.cow.CopyOnWriteInputFormat;
+import org.apache.hudi.source.format.mor.MergeOnReadInputFormat;
+import org.apache.hudi.source.format.mor.MergeOnReadInputSplit;
+import org.apache.hudi.source.format.mor.MergeOnReadTableState;
+import org.apache.hudi.util.AvroSchemaConverter;
+import org.apache.hudi.util.StreamerUtil;
+
+import org.apache.avro.Schema;
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.common.io.FileInputFormat;
+import org.apache.flink.api.common.io.FilePathFilter;
+import org.apache.flink.api.common.io.InputFormat;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.io.CollectionInputFormat;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.DataStreamSource;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.source.InputFormatSourceFunction;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.expressions.Expression;
+import org.apache.flink.table.runtime.types.TypeInfoDataTypeConverter;
+import org.apache.flink.table.sources.FilterableTableSource;
+import org.apache.flink.table.sources.LimitableTableSource;
+import org.apache.flink.table.sources.PartitionableTableSource;
+import org.apache.flink.table.sources.ProjectableTableSource;
+import org.apache.flink.table.sources.StreamTableSource;
+import org.apache.flink.table.sources.TableSource;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.mapred.JobConf;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import static org.apache.hudi.hadoop.utils.HoodieRealtimeRecordReaderUtils.getMaxCompactionMemoryInBytes;
+import static org.apache.hudi.source.format.FormatUtils.getParquetConf;
+
+/**
+ * Hoodie batch table source that always read the latest snapshot of the underneath table.
+ */
+public class HoodieTableSource implements
+ StreamTableSource<RowData>,
+ PartitionableTableSource,
+ ProjectableTableSource<RowData>,
+ LimitableTableSource<RowData>,
+ FilterableTableSource<RowData> {
+ private static final Logger LOG = LoggerFactory.getLogger(HoodieTableSource.class);
+
+ private final transient org.apache.hadoop.conf.Configuration hadoopConf;
+ private final transient HoodieTableMetaClient metaClient;
+ private final long maxCompactionMemoryInBytes;
+
+ private final TableSchema schema;
+ private final Path path;
+ private final List<String> partitionKeys;
+ private final String defaultPartName;
+ private final Configuration conf;
+
+ private final int[] requiredPos;
+ private final long limit;
+ private final List<Expression> filters;
+
+ private List<Map<String, String>> requiredPartitions;
+
+ public HoodieTableSource(
+ TableSchema schema,
+ Path path,
+ List<String> partitionKeys,
+ String defaultPartName,
+ Configuration conf) {
+ this(schema, path, partitionKeys, defaultPartName, conf,
+ null, null, null, null);
+ }
+
+ public HoodieTableSource(
+ TableSchema schema,
+ Path path,
+ List<String> partitionKeys,
+ String defaultPartName,
+ Configuration conf,
+ @Nullable List<Map<String, String>> requiredPartitions,
+ @Nullable int[] requiredPos,
+ @Nullable Long limit,
+ @Nullable List<Expression> filters) {
+ this.schema = schema;
+ this.path = path;
+ this.partitionKeys = partitionKeys;
+ this.defaultPartName = defaultPartName;
+ this.conf = conf;
+ this.requiredPartitions = requiredPartitions;
+ this.requiredPos = requiredPos == null
+ ? IntStream.range(0, schema.getFieldCount()).toArray()
+ : requiredPos;
+ this.limit = limit == null ? -1 : limit;
Review comment:
Let us extract the `-1` to be a more readable constant.
##########
File path: hudi-flink/src/main/java/org/apache/hudi/source/format/FilePathUtils.java
##########
@@ -0,0 +1,320 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.source.format;
+
+import org.apache.hudi.common.fs.FSUtils;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.core.fs.FileStatus;
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.utils.PartitionPathUtils;
+import org.apache.hadoop.conf.Configuration;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.BitSet;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+/**
+ * Reference the Flink {@link PartitionPathUtils} but supports simple partition path besides the Hive style.
+ */
+public class FilePathUtils {
+
+ private static final Pattern PARTITION_NAME_PATTERN = Pattern.compile("([^/]+)=([^/]+)");
Review comment:
Does this pattern match the hive partition style? If yes, it would be better to rename with `HIVE_STYLE_PARTITION_PATTERN` ?
##########
File path: hudi-flink/src/main/java/org/apache/hudi/source/format/FilePathUtils.java
##########
@@ -0,0 +1,320 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.source.format;
+
+import org.apache.hudi.common.fs.FSUtils;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.core.fs.FileStatus;
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.utils.PartitionPathUtils;
+import org.apache.hadoop.conf.Configuration;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.BitSet;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+/**
+ * Reference the Flink {@link PartitionPathUtils} but supports simple partition path besides the Hive style.
Review comment:
You mean `org.apache.flink.table.utils. PartitionPathUtils `? If yes, let us add the package name?
##########
File path: hudi-flink/src/main/java/org/apache/hudi/source/HoodieTableSource.java
##########
@@ -0,0 +1,411 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.source;
+
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.model.HoodieBaseFile;
+import org.apache.hudi.common.model.HoodieLogFile;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.TableSchemaResolver;
+import org.apache.hudi.common.table.view.HoodieTableFileSystemView;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.hadoop.HoodieROTablePathFilter;
+import org.apache.hudi.hadoop.utils.HoodieRealtimeInputFormatUtils;
+import org.apache.hudi.operator.FlinkOptions;
+import org.apache.hudi.source.format.FilePathUtils;
+import org.apache.hudi.source.format.cow.CopyOnWriteInputFormat;
+import org.apache.hudi.source.format.mor.MergeOnReadInputFormat;
+import org.apache.hudi.source.format.mor.MergeOnReadInputSplit;
+import org.apache.hudi.source.format.mor.MergeOnReadTableState;
+import org.apache.hudi.util.AvroSchemaConverter;
+import org.apache.hudi.util.StreamerUtil;
+
+import org.apache.avro.Schema;
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.common.io.FileInputFormat;
+import org.apache.flink.api.common.io.FilePathFilter;
+import org.apache.flink.api.common.io.InputFormat;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.io.CollectionInputFormat;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.DataStreamSource;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.source.InputFormatSourceFunction;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.expressions.Expression;
+import org.apache.flink.table.runtime.types.TypeInfoDataTypeConverter;
+import org.apache.flink.table.sources.FilterableTableSource;
+import org.apache.flink.table.sources.LimitableTableSource;
+import org.apache.flink.table.sources.PartitionableTableSource;
+import org.apache.flink.table.sources.ProjectableTableSource;
+import org.apache.flink.table.sources.StreamTableSource;
+import org.apache.flink.table.sources.TableSource;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.mapred.JobConf;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import static org.apache.hudi.hadoop.utils.HoodieRealtimeRecordReaderUtils.getMaxCompactionMemoryInBytes;
+import static org.apache.hudi.source.format.FormatUtils.getParquetConf;
+
+/**
+ * Hoodie batch table source that always read the latest snapshot of the underneath table.
+ */
+public class HoodieTableSource implements
+ StreamTableSource<RowData>,
+ PartitionableTableSource,
+ ProjectableTableSource<RowData>,
+ LimitableTableSource<RowData>,
+ FilterableTableSource<RowData> {
+ private static final Logger LOG = LoggerFactory.getLogger(HoodieTableSource.class);
+
+ private final transient org.apache.hadoop.conf.Configuration hadoopConf;
+ private final transient HoodieTableMetaClient metaClient;
+ private final long maxCompactionMemoryInBytes;
+
+ private final TableSchema schema;
+ private final Path path;
+ private final List<String> partitionKeys;
+ private final String defaultPartName;
+ private final Configuration conf;
+
+ private final int[] requiredPos;
+ private final long limit;
+ private final List<Expression> filters;
+
+ private List<Map<String, String>> requiredPartitions;
+
+ public HoodieTableSource(
+ TableSchema schema,
+ Path path,
+ List<String> partitionKeys,
+ String defaultPartName,
+ Configuration conf) {
+ this(schema, path, partitionKeys, defaultPartName, conf,
+ null, null, null, null);
+ }
+
+ public HoodieTableSource(
+ TableSchema schema,
+ Path path,
+ List<String> partitionKeys,
+ String defaultPartName,
+ Configuration conf,
+ @Nullable List<Map<String, String>> requiredPartitions,
+ @Nullable int[] requiredPos,
+ @Nullable Long limit,
+ @Nullable List<Expression> filters) {
+ this.schema = schema;
+ this.path = path;
+ this.partitionKeys = partitionKeys;
+ this.defaultPartName = defaultPartName;
+ this.conf = conf;
+ this.requiredPartitions = requiredPartitions;
+ this.requiredPos = requiredPos == null
+ ? IntStream.range(0, schema.getFieldCount()).toArray()
+ : requiredPos;
+ this.limit = limit == null ? -1 : limit;
+ this.filters = filters == null ? Collections.emptyList() : filters;
+ final String basePath = this.conf.getString(FlinkOptions.PATH);
+ this.hadoopConf = StreamerUtil.getHadoopConf();
+ this.metaClient = HoodieTableMetaClient.builder().setConf(hadoopConf).setBasePath(basePath).build();
+ this.maxCompactionMemoryInBytes = getMaxCompactionMemoryInBytes(new JobConf(this.hadoopConf));
+ }
+
+ @Override
+ public DataStream<RowData> getDataStream(StreamExecutionEnvironment execEnv) {
+ @SuppressWarnings("unchecked")
+ TypeInformation<RowData> typeInfo =
+ (TypeInformation<RowData>) TypeInfoDataTypeConverter.fromDataTypeToTypeInfo(getProducedDataType());
+ InputFormatSourceFunction<RowData> func = new InputFormatSourceFunction<>(getInputFormat(), typeInfo);
+ DataStreamSource<RowData> source = execEnv.addSource(func, explainSource(), typeInfo);
+ return source.name(explainSource());
+ }
+
+ @Override
+ public boolean isBounded() {
+ return true;
+ }
+
+ @Override
+ public TableSource<RowData> applyPredicate(List<Expression> predicates) {
+ return new HoodieTableSource(schema, path, partitionKeys, defaultPartName, conf,
+ requiredPartitions, requiredPos, limit, new ArrayList<>(predicates));
+ }
+
+ @Override
+ public boolean isFilterPushedDown() {
+ return this.filters != null && this.filters.size() > 0;
+ }
+
+ @Override
+ public boolean isLimitPushedDown() {
+ return this.limit != -1;
+ }
+
+ @Override
+ public TableSource<RowData> applyLimit(long limit) {
+ return new HoodieTableSource(schema, path, partitionKeys, defaultPartName, conf,
+ requiredPartitions, requiredPos, limit, filters);
+ }
+
+ @Override
+ public List<Map<String, String>> getPartitions() {
+ try {
+ return FilePathUtils
+ .searchPartKeyValueAndPaths(
+ path.getFileSystem(),
+ path,
+ conf.getBoolean(FlinkOptions.HIVE_STYLE_PARTITION),
+ partitionKeys.toArray(new String[0]))
+ .stream()
+ .map(tuple2 -> tuple2.f0)
+ .map(spec -> {
+ LinkedHashMap<String, String> ret = new LinkedHashMap<>();
+ spec.forEach((k, v) -> ret.put(k, defaultPartName.equals(v) ? null : v));
+ return ret;
+ })
+ .collect(Collectors.toList());
+ } catch (Exception e) {
+ throw new TableException("Fetch partitions fail.", e);
+ }
+ }
+
+ @Override
+ public TableSource applyPartitionPruning(List<Map<String, String>> requiredPartitions) {
+ return new HoodieTableSource(schema, path, partitionKeys, defaultPartName, conf,
+ requiredPartitions, requiredPos, limit, filters);
+ }
+
+ @Override
+ public TableSource<RowData> projectFields(int[] requiredPos) {
+ return new HoodieTableSource(schema, path, partitionKeys, defaultPartName, conf,
+ requiredPartitions, requiredPos, limit, filters);
+ }
+
+ @Override
+ public TableSchema getTableSchema() {
+ return schema;
+ }
+
+ @Override
+ public DataType getProducedDataType() {
+ String[] schemaFieldNames = this.schema.getFieldNames();
+ DataType[] schemaTypes = this.schema.getFieldDataTypes();
+
+ return DataTypes.ROW(Arrays.stream(this.requiredPos)
+ .mapToObj(i -> DataTypes.FIELD(schemaFieldNames[i], schemaTypes[i]))
+ .toArray(DataTypes.Field[]::new))
+ .bridgedTo(RowData.class);
+ }
+
+ private List<Map<String, String>> getOrFetchPartitions() {
+ if (requiredPartitions == null) {
+ requiredPartitions = getPartitions();
+ }
+ return requiredPartitions;
+ }
+
+ private List<MergeOnReadInputSplit> buildFileIndex(Path[] paths) {
+ FileStatus[] fileStatuses = Arrays.stream(paths)
+ .flatMap(path -> Arrays.stream(FilePathUtils.getFileStatusRecursivelyV2(path, 1, hadoopConf)))
+ .toArray(FileStatus[]::new);
+ if (fileStatuses.length == 0) {
+ throw new HoodieException("No files found for reading in user provided path.");
+ }
+
+ HoodieTableFileSystemView fsView = new HoodieTableFileSystemView(metaClient,
+ metaClient.getActiveTimeline().getCommitsTimeline()
+ .filterCompletedInstants(), fileStatuses);
+ List<HoodieBaseFile> latestFiles = fsView.getLatestBaseFiles().collect(Collectors.toList());
+ String latestCommit = fsView.getLastInstant().get().getTimestamp();
+ final String mergeType = this.conf.getString(FlinkOptions.MERGE_TYPE);
+ final AtomicInteger cnt = new AtomicInteger(0);
+ if (latestFiles.size() > 0) {
+ Map<HoodieBaseFile, List<String>> fileGroup =
+ HoodieRealtimeInputFormatUtils.groupLogsByBaseFile(hadoopConf, latestFiles);
+ return fileGroup.entrySet().stream().map(kv -> {
+ HoodieBaseFile baseFile = kv.getKey();
+ Option<List<String>> logPaths = kv.getValue().size() == 0
+ ? Option.empty()
+ : Option.of(kv.getValue());
+ return new MergeOnReadInputSplit(cnt.getAndAdd(1),
+ baseFile.getPath(), logPaths, latestCommit,
+ metaClient.getBasePath(), maxCompactionMemoryInBytes, mergeType);
+ }).collect(Collectors.toList());
+ } else {
+ // all the files are logs
+ return Arrays.stream(paths).map(partitionPath -> {
+ String relPartitionPath = FSUtils.getRelativePartitionPath(
+ new org.apache.hadoop.fs.Path(path.toUri()),
+ new org.apache.hadoop.fs.Path(partitionPath.toUri()));
+ return fsView.getLatestMergedFileSlicesBeforeOrOn(relPartitionPath, latestCommit)
+ .map(fileSlice -> {
+ Option<List<String>> logPaths = Option.ofNullable(fileSlice.getLogFiles()
+ .sorted(HoodieLogFile.getLogFileComparator())
+ .map(logFile -> logFile.getPath().toString())
+ .collect(Collectors.toList()));
+ return new MergeOnReadInputSplit(cnt.getAndAdd(1),
+ null, logPaths, latestCommit,
+ metaClient.getBasePath(), maxCompactionMemoryInBytes, mergeType);
+ }).collect(Collectors.toList()); })
+ .flatMap(Collection::stream)
+ .collect(Collectors.toList());
+ }
+ }
+
+ @VisibleForTesting
+ public InputFormat<RowData, ?> getInputFormat() {
+ // When this table has no partition, just return an empty source.
+ if (!partitionKeys.isEmpty() && getOrFetchPartitions().isEmpty()) {
+ return new CollectionInputFormat<>(Collections.emptyList(), null);
+ }
+
+ final Path[] paths = getReadPaths();
+ if (paths.length == 0) {
+ return new CollectionInputFormat<>(Collections.emptyList(), null);
+ }
+
+ TableSchemaResolver schemaUtil = new TableSchemaResolver(metaClient);
+ final Schema tableAvroSchema;
+ try {
+ tableAvroSchema = schemaUtil.getTableAvroSchema();
+ } catch (Exception e) {
+ throw new HoodieException("Get table avro schema error", e);
+ }
+ final DataType rowDataType = AvroSchemaConverter.convertToDataType(tableAvroSchema);
+ final RowType rowType = (RowType) rowDataType.getLogicalType();
+ final RowType requiredRowType = (RowType) getProducedDataType().notNull().getLogicalType();
+
+ final String queryType = this.conf.getString(FlinkOptions.QUERY_TYPE);
+ if (queryType.equals(FlinkOptions.QUERY_TYPE_SNAPSHOT)) {
+ switch (this.conf.getString(FlinkOptions.TABLE_TYPE)) {
+ case FlinkOptions.TABLE_TYPE_MERGE_ON_READ:
Review comment:
Why do not use the enum?
##########
File path: hudi-flink/src/main/java/org/apache/hudi/sink/HoodieTableSink.java
##########
@@ -0,0 +1,127 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.sink;
+
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.operator.FlinkOptions;
+import org.apache.hudi.operator.StreamWriteOperatorFactory;
+import org.apache.hudi.operator.compact.CompactFunction;
+import org.apache.hudi.operator.compact.CompactionCommitEvent;
+import org.apache.hudi.operator.compact.CompactionCommitSink;
+import org.apache.hudi.operator.compact.CompactionPlanEvent;
+import org.apache.hudi.operator.compact.CompactionPlanOperator;
+import org.apache.hudi.operator.partitioner.BucketAssignFunction;
+import org.apache.hudi.operator.transform.RowDataToHoodieFunction;
+import org.apache.hudi.util.StreamerUtil;
+
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.DataStreamSink;
+import org.apache.flink.streaming.api.functions.sink.SinkFunction;
+import org.apache.flink.streaming.api.operators.KeyedProcessOperator;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.sinks.AppendStreamTableSink;
+import org.apache.flink.table.sinks.PartitionableTableSink;
+import org.apache.flink.table.sinks.TableSink;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.RowType;
+
+import java.util.Map;
+
+/**
+ * Hoodie table sink.
+ */
+public class HoodieTableSink implements AppendStreamTableSink<RowData>, PartitionableTableSink {
+
+ private final Configuration conf;
+ private final TableSchema schema;
+ private final boolean isBounded;
+
+ public HoodieTableSink(Configuration conf, TableSchema schema, boolean isBounded) {
+ this.conf = conf;
+ this.schema = schema;
+ this.isBounded = isBounded;
+ }
+
+ @Override
+ public DataStreamSink<?> consumeDataStream(DataStream<RowData> dataStream) {
+ // Read from kafka source
+ RowType rowType = (RowType) this.schema.toRowDataType().notNull().getLogicalType();
+ int numWriteTasks = this.conf.getInteger(FlinkOptions.WRITE_TASK_PARALLELISM);
+ StreamWriteOperatorFactory<HoodieRecord> operatorFactory = new StreamWriteOperatorFactory<>(conf, isBounded);
+
+ DataStream<Object> pipeline = dataStream
+ // Key-by partition path, to avoid multiple subtasks write to a partition at the same time
Review comment:
duplicated comment?
##########
File path: hudi-flink/src/main/java/org/apache/hudi/source/HoodieTableSource.java
##########
@@ -0,0 +1,411 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.source;
+
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.model.HoodieBaseFile;
+import org.apache.hudi.common.model.HoodieLogFile;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.TableSchemaResolver;
+import org.apache.hudi.common.table.view.HoodieTableFileSystemView;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.hadoop.HoodieROTablePathFilter;
+import org.apache.hudi.hadoop.utils.HoodieRealtimeInputFormatUtils;
+import org.apache.hudi.operator.FlinkOptions;
+import org.apache.hudi.source.format.FilePathUtils;
+import org.apache.hudi.source.format.cow.CopyOnWriteInputFormat;
+import org.apache.hudi.source.format.mor.MergeOnReadInputFormat;
+import org.apache.hudi.source.format.mor.MergeOnReadInputSplit;
+import org.apache.hudi.source.format.mor.MergeOnReadTableState;
+import org.apache.hudi.util.AvroSchemaConverter;
+import org.apache.hudi.util.StreamerUtil;
+
+import org.apache.avro.Schema;
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.common.io.FileInputFormat;
+import org.apache.flink.api.common.io.FilePathFilter;
+import org.apache.flink.api.common.io.InputFormat;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.io.CollectionInputFormat;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.DataStreamSource;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.source.InputFormatSourceFunction;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.expressions.Expression;
+import org.apache.flink.table.runtime.types.TypeInfoDataTypeConverter;
+import org.apache.flink.table.sources.FilterableTableSource;
+import org.apache.flink.table.sources.LimitableTableSource;
+import org.apache.flink.table.sources.PartitionableTableSource;
+import org.apache.flink.table.sources.ProjectableTableSource;
+import org.apache.flink.table.sources.StreamTableSource;
+import org.apache.flink.table.sources.TableSource;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.mapred.JobConf;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import static org.apache.hudi.hadoop.utils.HoodieRealtimeRecordReaderUtils.getMaxCompactionMemoryInBytes;
+import static org.apache.hudi.source.format.FormatUtils.getParquetConf;
+
+/**
+ * Hoodie batch table source that always read the latest snapshot of the underneath table.
+ */
+public class HoodieTableSource implements
+ StreamTableSource<RowData>,
+ PartitionableTableSource,
+ ProjectableTableSource<RowData>,
+ LimitableTableSource<RowData>,
+ FilterableTableSource<RowData> {
+ private static final Logger LOG = LoggerFactory.getLogger(HoodieTableSource.class);
+
+ private final transient org.apache.hadoop.conf.Configuration hadoopConf;
+ private final transient HoodieTableMetaClient metaClient;
+ private final long maxCompactionMemoryInBytes;
+
+ private final TableSchema schema;
+ private final Path path;
+ private final List<String> partitionKeys;
+ private final String defaultPartName;
+ private final Configuration conf;
+
+ private final int[] requiredPos;
+ private final long limit;
+ private final List<Expression> filters;
+
+ private List<Map<String, String>> requiredPartitions;
+
+ public HoodieTableSource(
+ TableSchema schema,
+ Path path,
+ List<String> partitionKeys,
+ String defaultPartName,
+ Configuration conf) {
+ this(schema, path, partitionKeys, defaultPartName, conf,
+ null, null, null, null);
Review comment:
IMO, it's not necessary to break this new line.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] codecov-io edited a comment on pull request #2613: [HUDI-1647] Supports snapshot read for Flink
Posted by GitBox <gi...@apache.org>.
codecov-io edited a comment on pull request #2613:
URL: https://github.com/apache/hudi/pull/2613#issuecomment-787825622
# [Codecov](https://codecov.io/gh/apache/hudi/pull/2613?src=pr&el=h1) Report
> Merging [#2613](https://codecov.io/gh/apache/hudi/pull/2613?src=pr&el=desc) (c0a3083) into [master](https://codecov.io/gh/apache/hudi/commit/7a11de12764d8f68f296c6e68a22822318bfbefa?el=desc) (7a11de1) will **decrease** coverage by `0.15%`.
> The diff coverage is `47.23%`.
[](https://codecov.io/gh/apache/hudi/pull/2613?src=pr&el=tree)
```diff
@@ Coverage Diff @@
## master #2613 +/- ##
============================================
- Coverage 51.56% 51.40% -0.16%
- Complexity 3286 3479 +193
============================================
Files 445 461 +16
Lines 20328 21675 +1347
Branches 2102 2299 +197
============================================
+ Hits 10483 11143 +660
- Misses 8978 9572 +594
- Partials 867 960 +93
```
| Flag | Coverage Δ | Complexity Δ | |
|---|---|---|---|
| hudicli | `36.87% <ø> (ø)` | `0.00 <ø> (ø)` | |
| hudiclient | `100.00% <ø> (ø)` | `0.00 <ø> (ø)` | |
| hudicommon | `51.30% <ø> (-0.03%)` | `0.00 <ø> (ø)` | |
| hudiflink | `50.36% <47.23%> (-1.03%)` | `0.00 <188.00> (ø)` | |
| hudihadoopmr | `33.16% <ø> (ø)` | `0.00 <ø> (ø)` | |
| hudisparkdatasource | `69.71% <ø> (ø)` | `0.00 <ø> (ø)` | |
| hudisync | `49.62% <ø> (ø)` | `0.00 <ø> (ø)` | |
| huditimelineservice | `66.49% <ø> (ø)` | `0.00 <ø> (ø)` | |
| hudiutilities | `69.54% <ø> (ø)` | `0.00 <ø> (ø)` | |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags#carryforward-flags-in-the-pull-request-comment) to find out more.
| [Impacted Files](https://codecov.io/gh/apache/hudi/pull/2613?src=pr&el=tree) | Coverage Δ | Complexity Δ | |
|---|---|---|---|
| [...va/org/apache/hudi/factory/HoodieTableFactory.java](https://codecov.io/gh/apache/hudi/pull/2613/diff?src=pr&el=tree#diff-aHVkaS1mbGluay9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvaHVkaS9mYWN0b3J5L0hvb2RpZVRhYmxlRmFjdG9yeS5qYXZh) | `0.00% <0.00%> (ø)` | `0.00 <0.00> (?)` | |
| [.../org/apache/hudi/operator/StreamWriteFunction.java](https://codecov.io/gh/apache/hudi/pull/2613/diff?src=pr&el=tree#diff-aHVkaS1mbGluay9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvaHVkaS9vcGVyYXRvci9TdHJlYW1Xcml0ZUZ1bmN0aW9uLmphdmE=) | `84.00% <0.00%> (-2.60%)` | `22.00 <0.00> (ø)` | |
| [.../org/apache/hudi/operator/StreamWriteOperator.java](https://codecov.io/gh/apache/hudi/pull/2613/diff?src=pr&el=tree#diff-aHVkaS1mbGluay9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvaHVkaS9vcGVyYXRvci9TdHJlYW1Xcml0ZU9wZXJhdG9yLmphdmE=) | `0.00% <0.00%> (ø)` | `0.00 <0.00> (ø)` | |
| [...ache/hudi/operator/StreamWriteOperatorFactory.java](https://codecov.io/gh/apache/hudi/pull/2613/diff?src=pr&el=tree#diff-aHVkaS1mbGluay9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvaHVkaS9vcGVyYXRvci9TdHJlYW1Xcml0ZU9wZXJhdG9yRmFjdG9yeS5qYXZh) | `0.00% <0.00%> (ø)` | `0.00 <0.00> (ø)` | |
| [...udi/operator/partitioner/BucketAssignFunction.java](https://codecov.io/gh/apache/hudi/pull/2613/diff?src=pr&el=tree#diff-aHVkaS1mbGluay9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvaHVkaS9vcGVyYXRvci9wYXJ0aXRpb25lci9CdWNrZXRBc3NpZ25GdW5jdGlvbi5qYXZh) | `85.86% <ø> (+0.92%)` | `22.00 <0.00> (+1.00)` | |
| [...ain/java/org/apache/hudi/sink/HoodieTableSink.java](https://codecov.io/gh/apache/hudi/pull/2613/diff?src=pr&el=tree#diff-aHVkaS1mbGluay9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvaHVkaS9zaW5rL0hvb2RpZVRhYmxlU2luay5qYXZh) | `0.00% <0.00%> (ø)` | `0.00 <0.00> (?)` | |
| [...e/hudi/source/format/cow/ParquetDecimalVector.java](https://codecov.io/gh/apache/hudi/pull/2613/diff?src=pr&el=tree#diff-aHVkaS1mbGluay9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvaHVkaS9zb3VyY2UvZm9ybWF0L2Nvdy9QYXJxdWV0RGVjaW1hbFZlY3Rvci5qYXZh) | `0.00% <0.00%> (ø)` | `0.00 <0.00> (?)` | |
| [.../org/apache/hudi/util/AvroToRowDataConverters.java](https://codecov.io/gh/apache/hudi/pull/2613/diff?src=pr&el=tree#diff-aHVkaS1mbGluay9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvaHVkaS91dGlsL0F2cm9Ub1Jvd0RhdGFDb252ZXJ0ZXJzLmphdmE=) | `16.94% <16.94%> (ø)` | `12.00 <12.00> (?)` | |
| [...hudi/source/format/cow/ParquetSplitReaderUtil.java](https://codecov.io/gh/apache/hudi/pull/2613/diff?src=pr&el=tree#diff-aHVkaS1mbGluay9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvaHVkaS9zb3VyY2UvZm9ybWF0L2Nvdy9QYXJxdWV0U3BsaXRSZWFkZXJVdGlsLmphdmE=) | `20.00% <20.00%> (ø)` | `13.00 <13.00> (?)` | |
| [...java/org/apache/hudi/util/AvroSchemaConverter.java](https://codecov.io/gh/apache/hudi/pull/2613/diff?src=pr&el=tree#diff-aHVkaS1mbGluay9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvaHVkaS91dGlsL0F2cm9TY2hlbWFDb252ZXJ0ZXIuamF2YQ==) | `23.74% <25.97%> (+23.74%)` | `13.00 <7.00> (+13.00)` | |
| ... and [36 more](https://codecov.io/gh/apache/hudi/pull/2613/diff?src=pr&el=tree-more) | |
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] yanghua commented on pull request #2613: [HUDI-1647] Supports snapshot read for Flink
Posted by GitBox <gi...@apache.org>.
yanghua commented on pull request #2613:
URL: https://github.com/apache/hudi/pull/2613#issuecomment-789577004
@danny0405 Can you give a checklist about which files are copied from the other projects?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] danny0405 commented on a change in pull request #2613: [HUDI-1647] Supports snapshot read for Flink
Posted by GitBox <gi...@apache.org>.
danny0405 commented on a change in pull request #2613:
URL: https://github.com/apache/hudi/pull/2613#discussion_r586098644
##########
File path: hudi-flink/src/main/java/org/apache/hudi/operator/FlinkOptions.java
##########
@@ -73,11 +118,13 @@ private FlinkOptions() {
.noDefaultValue()
.withDescription("Table name to register to Hive metastore");
+ public static final String TABLE_TYPE_COPY_ON_WRITE = "COPY_ON_WRITE";
Review comment:
No, i think a constant here is more easy to use.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] danny0405 commented on pull request #2613: [HUDI-1647] Supports snapshot read for Flink
Posted by GitBox <gi...@apache.org>.
danny0405 commented on pull request #2613:
URL: https://github.com/apache/hudi/pull/2613#issuecomment-789583005
Sure
- AbstractColumnReader
- CopyOnWriteInputFormat
- ParquetColumnarRowSplitReader
- ParquetDecimalVector
- ParquetSplitReaderUtil
- RunLengthDecoder
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] garyli1019 commented on a change in pull request #2613: [HUDI-1647] Supports snapshot read for Flink
Posted by GitBox <gi...@apache.org>.
garyli1019 commented on a change in pull request #2613:
URL: https://github.com/apache/hudi/pull/2613#discussion_r585502738
##########
File path: hudi-flink/src/main/java/org/apache/hudi/sink/HoodieTableSink.java
##########
@@ -0,0 +1,126 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.sink;
+
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.operator.FlinkOptions;
+import org.apache.hudi.operator.StreamWriteOperatorFactory;
+import org.apache.hudi.operator.compact.CompactFunction;
+import org.apache.hudi.operator.compact.CompactionCommitEvent;
+import org.apache.hudi.operator.compact.CompactionCommitSink;
+import org.apache.hudi.operator.compact.CompactionPlanEvent;
+import org.apache.hudi.operator.compact.CompactionPlanOperator;
+import org.apache.hudi.operator.partitioner.BucketAssignFunction;
+import org.apache.hudi.operator.transform.RowDataToHoodieFunction;
+import org.apache.hudi.util.StreamerUtil;
+
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.DataStreamSink;
+import org.apache.flink.streaming.api.functions.sink.SinkFunction;
+import org.apache.flink.streaming.api.operators.KeyedProcessOperator;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.sinks.AppendStreamTableSink;
+import org.apache.flink.table.sinks.PartitionableTableSink;
+import org.apache.flink.table.sinks.TableSink;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.RowType;
+
+import java.util.Map;
+
+/**
+ * Hoodie table sink.
+ */
+public class HoodieTableSink implements AppendStreamTableSink<RowData>, PartitionableTableSink {
Review comment:
Is this for the batch job?
##########
File path: hudi-flink/src/main/java/org/apache/hudi/source/HoodieTableSource.java
##########
@@ -0,0 +1,410 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.source;
+
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.model.HoodieBaseFile;
+import org.apache.hudi.common.model.HoodieLogFile;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.TableSchemaResolver;
+import org.apache.hudi.common.table.view.HoodieTableFileSystemView;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.hadoop.HoodieROTablePathFilter;
+import org.apache.hudi.hadoop.utils.HoodieRealtimeInputFormatUtils;
+import org.apache.hudi.operator.FlinkOptions;
+import org.apache.hudi.source.format.FilePathUtils;
+import org.apache.hudi.source.format.cow.CopyOnWriteInputFormat;
+import org.apache.hudi.source.format.mor.MergeOnReadInputFormat;
+import org.apache.hudi.source.format.mor.MergeOnReadInputSplit;
+import org.apache.hudi.source.format.mor.MergeOnReadTableState;
+import org.apache.hudi.util.AvroSchemaConverter;
+import org.apache.hudi.util.StreamerUtil;
+
+import org.apache.avro.Schema;
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.common.io.FileInputFormat;
+import org.apache.flink.api.common.io.FilePathFilter;
+import org.apache.flink.api.common.io.InputFormat;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.io.CollectionInputFormat;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.DataStreamSource;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.source.InputFormatSourceFunction;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.expressions.Expression;
+import org.apache.flink.table.runtime.types.TypeInfoDataTypeConverter;
+import org.apache.flink.table.sources.FilterableTableSource;
+import org.apache.flink.table.sources.LimitableTableSource;
+import org.apache.flink.table.sources.PartitionableTableSource;
+import org.apache.flink.table.sources.ProjectableTableSource;
+import org.apache.flink.table.sources.StreamTableSource;
+import org.apache.flink.table.sources.TableSource;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.mapred.JobConf;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import static org.apache.hudi.hadoop.utils.HoodieRealtimeRecordReaderUtils.getMaxCompactionMemoryInBytes;
+import static org.apache.hudi.source.format.FormatUtils.getParquetConf;
+
+/**
+ * Hoodie batch table source that always read the latest snapshot of the underneath table.
+ */
+public class HoodieTableSource implements
+ StreamTableSource<RowData>,
+ PartitionableTableSource,
+ ProjectableTableSource<RowData>,
+ LimitableTableSource<RowData>,
+ FilterableTableSource<RowData> {
+ private static final Logger LOG = LoggerFactory.getLogger(HoodieTableSource.class);
+
+ private final transient org.apache.hadoop.conf.Configuration hadoopConf;
+ private final transient HoodieTableMetaClient metaClient;
+ private final long maxCompactionMemoryInBytes;
+
+ private final TableSchema schema;
+ private final Path path;
+ private final List<String> partitionKeys;
+ private final String defaultPartName;
+ private final Configuration conf;
+
+ private final int[] requiredPos;
+ private final long limit;
+ private final List<Expression> filters;
+
+ private List<Map<String, String>> requiredPartitions;
+
+ public HoodieTableSource(
+ TableSchema schema,
Review comment:
Does the user has to define the schema when using this? In Spark, we use the TableSchemaResolver to get the latest schema from the FileSystem
##########
File path: hudi-flink/src/main/java/org/apache/hudi/source/format/cow/AbstractColumnReader.java
##########
@@ -0,0 +1,324 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.source.format.cow;
+
+import org.apache.flink.formats.parquet.vector.ParquetDictionary;
+import org.apache.flink.formats.parquet.vector.reader.ColumnReader;
+import org.apache.flink.table.data.vector.writable.WritableColumnVector;
+import org.apache.flink.table.data.vector.writable.WritableIntVector;
+import org.apache.parquet.Preconditions;
+import org.apache.parquet.bytes.ByteBufferInputStream;
+import org.apache.parquet.bytes.BytesInput;
+import org.apache.parquet.bytes.BytesUtils;
+import org.apache.parquet.column.ColumnDescriptor;
+import org.apache.parquet.column.Dictionary;
+import org.apache.parquet.column.Encoding;
+import org.apache.parquet.column.page.DataPage;
+import org.apache.parquet.column.page.DataPageV1;
+import org.apache.parquet.column.page.DataPageV2;
+import org.apache.parquet.column.page.DictionaryPage;
+import org.apache.parquet.column.page.PageReader;
+import org.apache.parquet.column.values.ValuesReader;
+import org.apache.parquet.io.ParquetDecodingException;
+import org.apache.parquet.schema.PrimitiveType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+
+import static org.apache.parquet.column.ValuesType.REPETITION_LEVEL;
+
+/**
+ * Abstract {@link ColumnReader}.
+ * See {@link org.apache.parquet.column.impl.ColumnReaderImpl},
+ * part of the code is referred from Apache Spark and Apache Parquet.
+ *
+ * <p>Note: Reference Flink release 1.11.2 {@link org.apache.flink.formats.parquet.vector.reader.AbstractColumnReader}
+ * because some of the package scope methods.
+ */
+public abstract class AbstractColumnReader<V extends WritableColumnVector>
Review comment:
Why do we need a separate reader. Does Flink have its own one for parquet?
##########
File path: hudi-flink/src/main/java/org/apache/hudi/operator/FlinkOptions.java
##########
@@ -73,11 +118,13 @@ private FlinkOptions() {
.noDefaultValue()
.withDescription("Table name to register to Hive metastore");
+ public static final String TABLE_TYPE_COPY_ON_WRITE = "COPY_ON_WRITE";
Review comment:
can we link the String to `HoodieTableType`?
##########
File path: hudi-flink/src/main/java/org/apache/hudi/source/HoodieTableSource.java
##########
@@ -0,0 +1,411 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.source;
+
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.model.HoodieBaseFile;
+import org.apache.hudi.common.model.HoodieLogFile;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.TableSchemaResolver;
+import org.apache.hudi.common.table.view.HoodieTableFileSystemView;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.hadoop.HoodieROTablePathFilter;
+import org.apache.hudi.hadoop.utils.HoodieRealtimeInputFormatUtils;
+import org.apache.hudi.operator.FlinkOptions;
+import org.apache.hudi.source.format.FilePathUtils;
+import org.apache.hudi.source.format.cow.CopyOnWriteInputFormat;
+import org.apache.hudi.source.format.mor.MergeOnReadInputFormat;
+import org.apache.hudi.source.format.mor.MergeOnReadInputSplit;
+import org.apache.hudi.source.format.mor.MergeOnReadTableState;
+import org.apache.hudi.util.AvroSchemaConverter;
+import org.apache.hudi.util.StreamerUtil;
+
+import org.apache.avro.Schema;
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.common.io.FileInputFormat;
+import org.apache.flink.api.common.io.FilePathFilter;
+import org.apache.flink.api.common.io.InputFormat;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.io.CollectionInputFormat;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.DataStreamSource;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.source.InputFormatSourceFunction;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.expressions.Expression;
+import org.apache.flink.table.runtime.types.TypeInfoDataTypeConverter;
+import org.apache.flink.table.sources.FilterableTableSource;
+import org.apache.flink.table.sources.LimitableTableSource;
+import org.apache.flink.table.sources.PartitionableTableSource;
+import org.apache.flink.table.sources.ProjectableTableSource;
+import org.apache.flink.table.sources.StreamTableSource;
+import org.apache.flink.table.sources.TableSource;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.mapred.JobConf;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import static org.apache.hudi.hadoop.utils.HoodieRealtimeRecordReaderUtils.getMaxCompactionMemoryInBytes;
+import static org.apache.hudi.source.format.FormatUtils.getParquetConf;
+
+/**
+ * Hoodie batch table source that always read the latest snapshot of the underneath table.
+ */
+public class HoodieTableSource implements
+ StreamTableSource<RowData>,
+ PartitionableTableSource,
+ ProjectableTableSource<RowData>,
+ LimitableTableSource<RowData>,
+ FilterableTableSource<RowData> {
+ private static final Logger LOG = LoggerFactory.getLogger(HoodieTableSource.class);
+
+ private static final int NO_LIMIT_CONSTANT = -1;
+
+ private final transient org.apache.hadoop.conf.Configuration hadoopConf;
+ private final transient HoodieTableMetaClient metaClient;
+ private final long maxCompactionMemoryInBytes;
+
+ private final TableSchema schema;
+ private final Path path;
+ private final List<String> partitionKeys;
+ private final String defaultPartName;
+ private final Configuration conf;
+
+ private final int[] requiredPos;
+ private final long limit;
+ private final List<Expression> filters;
+
+ private List<Map<String, String>> requiredPartitions;
+
+ public HoodieTableSource(
+ TableSchema schema,
+ Path path,
+ List<String> partitionKeys,
+ String defaultPartName,
+ Configuration conf) {
+ this(schema, path, partitionKeys, defaultPartName, conf, null, null, null, null);
+ }
+
+ public HoodieTableSource(
+ TableSchema schema,
+ Path path,
+ List<String> partitionKeys,
+ String defaultPartName,
+ Configuration conf,
+ @Nullable List<Map<String, String>> requiredPartitions,
+ @Nullable int[] requiredPos,
+ @Nullable Long limit,
+ @Nullable List<Expression> filters) {
+ this.schema = schema;
+ this.path = path;
+ this.partitionKeys = partitionKeys;
+ this.defaultPartName = defaultPartName;
+ this.conf = conf;
+ this.requiredPartitions = requiredPartitions;
+ this.requiredPos = requiredPos == null
+ ? IntStream.range(0, schema.getFieldCount()).toArray()
+ : requiredPos;
+ this.limit = limit == null ? NO_LIMIT_CONSTANT : limit;
+ this.filters = filters == null ? Collections.emptyList() : filters;
+ final String basePath = this.conf.getString(FlinkOptions.PATH);
+ this.hadoopConf = StreamerUtil.getHadoopConf();
+ this.metaClient = HoodieTableMetaClient.builder().setConf(hadoopConf).setBasePath(basePath).build();
+ this.maxCompactionMemoryInBytes = getMaxCompactionMemoryInBytes(new JobConf(this.hadoopConf));
+ }
+
+ @Override
+ public DataStream<RowData> getDataStream(StreamExecutionEnvironment execEnv) {
+ @SuppressWarnings("unchecked")
+ TypeInformation<RowData> typeInfo =
+ (TypeInformation<RowData>) TypeInfoDataTypeConverter.fromDataTypeToTypeInfo(getProducedDataType());
+ InputFormatSourceFunction<RowData> func = new InputFormatSourceFunction<>(getInputFormat(), typeInfo);
+ DataStreamSource<RowData> source = execEnv.addSource(func, explainSource(), typeInfo);
+ return source.name(explainSource());
+ }
+
+ @Override
+ public boolean isBounded() {
+ return true;
+ }
+
+ @Override
+ public TableSource<RowData> applyPredicate(List<Expression> predicates) {
+ return new HoodieTableSource(schema, path, partitionKeys, defaultPartName, conf,
+ requiredPartitions, requiredPos, limit, new ArrayList<>(predicates));
+ }
+
+ @Override
+ public boolean isFilterPushedDown() {
+ return this.filters != null && this.filters.size() > 0;
+ }
+
+ @Override
+ public boolean isLimitPushedDown() {
+ return this.limit != NO_LIMIT_CONSTANT;
+ }
+
+ @Override
+ public TableSource<RowData> applyLimit(long limit) {
+ return new HoodieTableSource(schema, path, partitionKeys, defaultPartName, conf,
+ requiredPartitions, requiredPos, limit, filters);
+ }
+
+ @Override
+ public List<Map<String, String>> getPartitions() {
+ try {
+ return FilePathUtils
+ .searchPartKeyValueAndPaths(
+ path.getFileSystem(),
+ path,
+ conf.getBoolean(FlinkOptions.HIVE_STYLE_PARTITION),
+ partitionKeys.toArray(new String[0]))
+ .stream()
+ .map(tuple2 -> tuple2.f0)
+ .map(spec -> {
+ LinkedHashMap<String, String> ret = new LinkedHashMap<>();
+ spec.forEach((k, v) -> ret.put(k, defaultPartName.equals(v) ? null : v));
+ return ret;
+ })
+ .collect(Collectors.toList());
+ } catch (Exception e) {
+ throw new TableException("Fetch partitions fail.", e);
+ }
+ }
+
+ @Override
+ public TableSource applyPartitionPruning(List<Map<String, String>> requiredPartitions) {
+ return new HoodieTableSource(schema, path, partitionKeys, defaultPartName, conf,
+ requiredPartitions, requiredPos, limit, filters);
+ }
+
+ @Override
+ public TableSource<RowData> projectFields(int[] requiredPos) {
+ return new HoodieTableSource(schema, path, partitionKeys, defaultPartName, conf,
+ requiredPartitions, requiredPos, limit, filters);
+ }
+
+ @Override
+ public TableSchema getTableSchema() {
+ return schema;
+ }
+
+ @Override
+ public DataType getProducedDataType() {
+ String[] schemaFieldNames = this.schema.getFieldNames();
+ DataType[] schemaTypes = this.schema.getFieldDataTypes();
+
+ return DataTypes.ROW(Arrays.stream(this.requiredPos)
+ .mapToObj(i -> DataTypes.FIELD(schemaFieldNames[i], schemaTypes[i]))
+ .toArray(DataTypes.Field[]::new))
+ .bridgedTo(RowData.class);
+ }
+
+ private List<Map<String, String>> getOrFetchPartitions() {
+ if (requiredPartitions == null) {
+ requiredPartitions = getPartitions();
+ }
+ return requiredPartitions;
+ }
+
+ private List<MergeOnReadInputSplit> buildFileIndex(Path[] paths) {
+ FileStatus[] fileStatuses = Arrays.stream(paths)
+ .flatMap(path -> Arrays.stream(FilePathUtils.getFileStatusRecursivelyV2(path, 1, hadoopConf)))
+ .toArray(FileStatus[]::new);
+ if (fileStatuses.length == 0) {
+ throw new HoodieException("No files found for reading in user provided path.");
+ }
+
+ HoodieTableFileSystemView fsView = new HoodieTableFileSystemView(metaClient,
+ metaClient.getActiveTimeline().getCommitsTimeline()
+ .filterCompletedInstants(), fileStatuses);
+ List<HoodieBaseFile> latestFiles = fsView.getLatestBaseFiles().collect(Collectors.toList());
+ String latestCommit = fsView.getLastInstant().get().getTimestamp();
+ final String mergeType = this.conf.getString(FlinkOptions.MERGE_TYPE);
+ final AtomicInteger cnt = new AtomicInteger(0);
+ if (latestFiles.size() > 0) {
+ Map<HoodieBaseFile, List<String>> fileGroup =
+ HoodieRealtimeInputFormatUtils.groupLogsByBaseFile(hadoopConf, latestFiles);
+ return fileGroup.entrySet().stream().map(kv -> {
+ HoodieBaseFile baseFile = kv.getKey();
+ Option<List<String>> logPaths = kv.getValue().size() == 0
+ ? Option.empty()
+ : Option.of(kv.getValue());
+ return new MergeOnReadInputSplit(cnt.getAndAdd(1),
+ baseFile.getPath(), logPaths, latestCommit,
+ metaClient.getBasePath(), maxCompactionMemoryInBytes, mergeType);
+ }).collect(Collectors.toList());
+ } else {
+ // all the files are logs
Review comment:
how is this possible?
##########
File path: hudi-flink/src/main/java/org/apache/hudi/source/format/FormatUtils.java
##########
@@ -0,0 +1,98 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.source.format;
+
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.table.log.HoodieMergedLogRecordScanner;
+import org.apache.hudi.hadoop.config.HoodieRealtimeConfig;
+import org.apache.hudi.operator.FlinkOptions;
+import org.apache.hudi.source.format.mor.MergeOnReadInputSplit;
+
+import org.apache.avro.Schema;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.avro.generic.GenericRecordBuilder;
+import org.apache.avro.generic.IndexedRecord;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+
+import java.util.Arrays;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+
+/**
+ * Utilities for format.
+ */
+public class FormatUtils {
+ private FormatUtils() {
+ }
+
+ public static GenericRecord buildAvroRecordBySchema(
+ IndexedRecord record,
+ Schema requiredSchema,
+ int[] requiredPos,
+ GenericRecordBuilder recordBuilder) {
+ List<Schema.Field> requiredFields = requiredSchema.getFields();
+ assert (requiredFields.size() == requiredPos.length);
+ Iterator<Integer> positionIterator = Arrays.stream(requiredPos).iterator();
+ requiredFields.forEach(f -> recordBuilder.set(f, record.get(positionIterator.next())));
+ return recordBuilder.build();
+ }
+
+ public static HoodieMergedLogRecordScanner scanLog(
+ MergeOnReadInputSplit split,
+ Schema logSchema,
+ Configuration config) {
+ FileSystem fs = FSUtils.getFs(split.getTablePath(), config);
+ return HoodieMergedLogRecordScanner.newBuilder()
Review comment:
duplicate with spark code, maybe we can move this to a shared place?
##########
File path: hudi-flink/src/main/java/org/apache/hudi/source/format/cow/CopyOnWriteInputFormat.java
##########
@@ -0,0 +1,134 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.source.format.cow;
+
+import org.apache.flink.api.common.io.FileInputFormat;
+import org.apache.flink.core.fs.FileInputSplit;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.formats.parquet.utils.SerializableConfiguration;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.utils.PartitionPathUtils;
+import org.apache.hadoop.conf.Configuration;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.LinkedHashMap;
+import java.util.List;
+
+import static org.apache.flink.table.data.vector.VectorizedColumnBatch.DEFAULT_SIZE;
+import static org.apache.flink.table.filesystem.RowPartitionComputer.restorePartValueFromType;
+
+/**
+ * An implementation of {@link FileInputFormat} to read {@link RowData} records
+ * from Parquet files.
+ *
+ * <p>Note: Reference Flink release 1.11.2
+ * {@code org.apache.flink.formats.parquet.ParquetFileSystemFormatFactory.ParquetInputFormat}
+ * to support TIMESTAMP_MILLIS.
+ *
+ * @see ParquetSplitReaderUtil
+ */
+public class CopyOnWriteInputFormat extends FileInputFormat<RowData> {
Review comment:
Why not extends from `ParquetInputFormat`?
##########
File path: hudi-flink/src/main/java/org/apache/hudi/source/format/cow/ParquetColumnarRowSplitReader.java
##########
@@ -0,0 +1,370 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.source.format.cow;
+
+import org.apache.flink.formats.parquet.vector.reader.ColumnReader;
+import org.apache.flink.table.data.ColumnarRowData;
+import org.apache.flink.table.data.vector.ColumnVector;
+import org.apache.flink.table.data.vector.VectorizedColumnBatch;
+import org.apache.flink.table.data.vector.writable.WritableColumnVector;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
+import org.apache.flink.util.FlinkRuntimeException;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.parquet.column.ColumnDescriptor;
+import org.apache.parquet.column.page.PageReadStore;
+import org.apache.parquet.filter2.compat.FilterCompat;
+import org.apache.parquet.hadoop.ParquetFileReader;
+import org.apache.parquet.hadoop.metadata.BlockMetaData;
+import org.apache.parquet.hadoop.metadata.ParquetMetadata;
+import org.apache.parquet.schema.GroupType;
+import org.apache.parquet.schema.MessageType;
+import org.apache.parquet.schema.Type;
+import org.apache.parquet.schema.Types;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+
+import static org.apache.hudi.source.format.cow.ParquetSplitReaderUtil.createColumnReader;
+import static org.apache.hudi.source.format.cow.ParquetSplitReaderUtil.createWritableColumnVector;
+import static org.apache.parquet.filter2.compat.RowGroupFilter.filterRowGroups;
+import static org.apache.parquet.format.converter.ParquetMetadataConverter.range;
+import static org.apache.parquet.hadoop.ParquetFileReader.readFooter;
+import static org.apache.parquet.hadoop.ParquetInputFormat.getFilter;
+
+/**
+ * This reader is used to read a {@link VectorizedColumnBatch} from input split.
+ *
+ * <p>Note: Reference Flink release 1.11.2
+ * {@code org.apache.flink.formats.parquet.vector.ParquetColumnarRowSplitReader}
+ * because it is package scope.
+ */
+public class ParquetColumnarRowSplitReader implements Closeable {
Review comment:
IMO we put too much dependency on the File Reader. It will be hard to maintain over time. I am more inclined to reuse the reader from the compute engine side. WDYT?
##########
File path: hudi-flink/src/main/java/org/apache/hudi/source/format/FilePathUtils.java
##########
@@ -0,0 +1,320 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.source.format;
+
+import org.apache.hudi.common.fs.FSUtils;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.core.fs.FileStatus;
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.table.api.TableException;
+import org.apache.hadoop.conf.Configuration;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.BitSet;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+/**
+ * Reference the Flink {@link org.apache.flink.table.utils.PartitionPathUtils}
+ * but supports simple partition path besides the Hive style.
+ */
+public class FilePathUtils {
+
+ private static final Pattern HIVE_PARTITION_NAME_PATTERN = Pattern.compile("([^/]+)=([^/]+)");
+
+ private static final BitSet CHAR_TO_ESCAPE = new BitSet(128);
+
+ static {
+ for (char c = 0; c < ' '; c++) {
+ CHAR_TO_ESCAPE.set(c);
+ }
+
+ /*
+ * ASCII 01-1F are HTTP control characters that need to be escaped.
+ * \u000A and \u000D are \n and \r, respectively.
+ */
+ char[] clist = new char[] {'\u0001', '\u0002', '\u0003', '\u0004',
+ '\u0005', '\u0006', '\u0007', '\u0008', '\u0009', '\n', '\u000B',
+ '\u000C', '\r', '\u000E', '\u000F', '\u0010', '\u0011', '\u0012',
+ '\u0013', '\u0014', '\u0015', '\u0016', '\u0017', '\u0018', '\u0019',
+ '\u001A', '\u001B', '\u001C', '\u001D', '\u001E', '\u001F',
+ '"', '#', '%', '\'', '*', '/', ':', '=', '?', '\\', '\u007F', '{',
+ '[', ']', '^'};
+
+ for (char c : clist) {
+ CHAR_TO_ESCAPE.set(c);
+ }
+ }
+
+ private static boolean needsEscaping(char c) {
+ return c < CHAR_TO_ESCAPE.size() && CHAR_TO_ESCAPE.get(c);
+ }
+
+ /**
+ * Make partition path from partition spec.
+ *
+ * @param partitionKVs The partition key value mapping
+ * @param hiveStylePartition Whether the partition path is with Hive style,
+ * e.g. {partition key} = {partition value}
+ * @return an escaped, valid partition name
+ */
+ public static String generatePartitionPath(
+ LinkedHashMap<String, String> partitionKVs,
+ boolean hiveStylePartition) {
+ if (partitionKVs.isEmpty()) {
+ return "";
+ }
+ StringBuilder suffixBuf = new StringBuilder();
+ int i = 0;
+ for (Map.Entry<String, String> e : partitionKVs.entrySet()) {
+ if (i > 0) {
+ suffixBuf.append(Path.SEPARATOR);
+ }
+ if (hiveStylePartition) {
+ suffixBuf.append(escapePathName(e.getKey()));
+ suffixBuf.append('=');
+ }
+ suffixBuf.append(escapePathName(e.getValue()));
+ i++;
+ }
+ suffixBuf.append(Path.SEPARATOR);
+ return suffixBuf.toString();
+ }
+
+ /**
+ * Escapes a path name.
+ *
+ * @param path The path to escape.
+ * @return An escaped path name.
+ */
+ private static String escapePathName(String path) {
+ if (path == null || path.length() == 0) {
+ throw new TableException("Path should not be null or empty: " + path);
+ }
+
+ StringBuilder sb = new StringBuilder();
+ for (int i = 0; i < path.length(); i++) {
+ char c = path.charAt(i);
+ if (needsEscaping(c)) {
+ sb.append('%');
+ sb.append(String.format("%1$02X", (int) c));
+ } else {
+ sb.append(c);
+ }
+ }
+ return sb.toString();
+ }
+
+ /**
+ * Generates partition values from path.
+ *
+ * @param currPath Partition file path
+ * @param hivePartition Whether the partition path is with Hive style
+ * @param partitionKeys Partition keys
+ * @return Sequential partition specs.
+ */
+ public static List<String> extractPartitionValues(
+ Path currPath,
+ boolean hivePartition,
+ String[] partitionKeys) {
+ return new ArrayList<>(extractPartitionKeyValues(currPath, hivePartition, partitionKeys).values());
+ }
+
+ /**
+ * Generates partition key value mapping from path.
+ *
+ * @param currPath Partition file path
+ * @param hivePartition Whether the partition path is with Hive style
+ * @param partitionKeys Partition keys
+ * @return Sequential partition specs.
+ */
+ public static LinkedHashMap<String, String> extractPartitionKeyValues(
+ Path currPath,
+ boolean hivePartition,
+ String[] partitionKeys) {
+ LinkedHashMap<String, String> fullPartSpec = new LinkedHashMap<>();
+ List<String[]> kvs = new ArrayList<>();
+ int curDepth = 0;
+ do {
+ String component = currPath.getName();
+ final String[] kv = new String[2];
+ if (hivePartition) {
+ Matcher m = HIVE_PARTITION_NAME_PATTERN.matcher(component);
+ if (m.matches()) {
+ String k = unescapePathName(m.group(1));
+ String v = unescapePathName(m.group(2));
+ kv[0] = k;
+ kv[1] = v;
+ }
+ } else {
+ kv[0] = partitionKeys[partitionKeys.length - 1 - curDepth];
+ kv[1] = unescapePathName(component);
+ }
+ kvs.add(kv);
+ currPath = currPath.getParent();
+ curDepth++;
+ } while (currPath != null && !currPath.getName().isEmpty() && curDepth < partitionKeys.length);
+
+ // reverse the list since we checked the part from leaf dir to table's base dir
+ for (int i = kvs.size(); i > 0; i--) {
+ fullPartSpec.put(kvs.get(i - 1)[0], kvs.get(i - 1)[1]);
+ }
+
+ return fullPartSpec;
+ }
+
+ public static String unescapePathName(String path) {
+ StringBuilder sb = new StringBuilder();
+ for (int i = 0; i < path.length(); i++) {
+ char c = path.charAt(i);
+ if (c == '%' && i + 2 < path.length()) {
+ int code = -1;
+ try {
+ code = Integer.parseInt(path.substring(i + 1, i + 3), 16);
+ } catch (Exception ignored) {
+ // do nothing
+ }
+ if (code >= 0) {
+ sb.append((char) code);
+ i += 2;
+ continue;
+ }
+ }
+ sb.append(c);
+ }
+ return sb.toString();
+ }
+
+ /**
+ * Search all partitions in this path.
+ *
+ * @param fs File system
+ * @param path Search path
+ * @param hivePartition Whether the partition path is with Hive style
+ * @param partitionKeys Partition keys
+ * @return all partition key value mapping in sequence of the given path
+ */
+ public static List<Tuple2<LinkedHashMap<String, String>, Path>> searchPartKeyValueAndPaths(
+ FileSystem fs,
+ Path path,
+ boolean hivePartition,
+ String[] partitionKeys) {
+ // expectLevel start from 0, E.G. base_path/level0/level1/level2
+ FileStatus[] generatedParts = getFileStatusRecursively(path, partitionKeys.length, fs);
+ List<Tuple2<LinkedHashMap<String, String>, Path>> ret = new ArrayList<>();
+ for (FileStatus part : generatedParts) {
+ ret.add(
+ new Tuple2<>(
+ extractPartitionKeyValues(part.getPath(), hivePartition, partitionKeys),
+ part.getPath()));
+ }
+ return ret;
+ }
+
+ private static FileStatus[] getFileStatusRecursively(Path path, int expectLevel, FileSystem fs) {
+ ArrayList<FileStatus> result = new ArrayList<>();
+
+ try {
+ FileStatus fileStatus = fs.getFileStatus(path);
+ listStatusRecursively(fs, fileStatus, 0, expectLevel, result);
+ } catch (IOException ignore) {
+ return new FileStatus[0];
+ }
+
+ return result.toArray(new FileStatus[0]);
+ }
+
+ private static void listStatusRecursively(
+ FileSystem fs,
+ FileStatus fileStatus,
+ int level,
+ int expectLevel,
+ List<FileStatus> results) throws IOException {
+ if (expectLevel == level && !isHiddenFile(fileStatus)) {
+ results.add(fileStatus);
+ return;
+ }
+
+ if (fileStatus.isDir() && !isHiddenFile(fileStatus)) {
+ for (FileStatus stat : fs.listStatus(fileStatus.getPath())) {
+ listStatusRecursively(fs, stat, level + 1, expectLevel, results);
+ }
+ }
+ }
+
+ private static boolean isHiddenFile(FileStatus fileStatus) {
+ String name = fileStatus.getPath().getName();
+ return name.startsWith("_") || name.startsWith(".");
+ }
+
+ /**
+ * Same as getFileStatusRecursively but returns hadoop {@link org.apache.hadoop.fs.FileStatus}s.
+ */
+ public static org.apache.hadoop.fs.FileStatus[] getFileStatusRecursivelyV2(
Review comment:
how about `getHadoopFileStatusRecursively`
##########
File path: hudi-flink/src/main/java/org/apache/hudi/source/format/FilePathUtils.java
##########
@@ -0,0 +1,320 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.source.format;
+
+import org.apache.hudi.common.fs.FSUtils;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.core.fs.FileStatus;
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.table.api.TableException;
+import org.apache.hadoop.conf.Configuration;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.BitSet;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+/**
+ * Reference the Flink {@link org.apache.flink.table.utils.PartitionPathUtils}
+ * but supports simple partition path besides the Hive style.
+ */
+public class FilePathUtils {
+
+ private static final Pattern HIVE_PARTITION_NAME_PATTERN = Pattern.compile("([^/]+)=([^/]+)");
+
+ private static final BitSet CHAR_TO_ESCAPE = new BitSet(128);
+
+ static {
+ for (char c = 0; c < ' '; c++) {
+ CHAR_TO_ESCAPE.set(c);
+ }
+
+ /*
+ * ASCII 01-1F are HTTP control characters that need to be escaped.
+ * \u000A and \u000D are \n and \r, respectively.
+ */
+ char[] clist = new char[] {'\u0001', '\u0002', '\u0003', '\u0004',
+ '\u0005', '\u0006', '\u0007', '\u0008', '\u0009', '\n', '\u000B',
+ '\u000C', '\r', '\u000E', '\u000F', '\u0010', '\u0011', '\u0012',
+ '\u0013', '\u0014', '\u0015', '\u0016', '\u0017', '\u0018', '\u0019',
+ '\u001A', '\u001B', '\u001C', '\u001D', '\u001E', '\u001F',
+ '"', '#', '%', '\'', '*', '/', ':', '=', '?', '\\', '\u007F', '{',
+ '[', ']', '^'};
+
+ for (char c : clist) {
+ CHAR_TO_ESCAPE.set(c);
+ }
+ }
+
+ private static boolean needsEscaping(char c) {
+ return c < CHAR_TO_ESCAPE.size() && CHAR_TO_ESCAPE.get(c);
+ }
+
+ /**
+ * Make partition path from partition spec.
+ *
+ * @param partitionKVs The partition key value mapping
+ * @param hiveStylePartition Whether the partition path is with Hive style,
+ * e.g. {partition key} = {partition value}
+ * @return an escaped, valid partition name
+ */
+ public static String generatePartitionPath(
+ LinkedHashMap<String, String> partitionKVs,
+ boolean hiveStylePartition) {
+ if (partitionKVs.isEmpty()) {
+ return "";
+ }
+ StringBuilder suffixBuf = new StringBuilder();
+ int i = 0;
+ for (Map.Entry<String, String> e : partitionKVs.entrySet()) {
+ if (i > 0) {
+ suffixBuf.append(Path.SEPARATOR);
+ }
+ if (hiveStylePartition) {
+ suffixBuf.append(escapePathName(e.getKey()));
+ suffixBuf.append('=');
+ }
+ suffixBuf.append(escapePathName(e.getValue()));
+ i++;
+ }
+ suffixBuf.append(Path.SEPARATOR);
+ return suffixBuf.toString();
+ }
+
+ /**
+ * Escapes a path name.
+ *
+ * @param path The path to escape.
+ * @return An escaped path name.
+ */
+ private static String escapePathName(String path) {
+ if (path == null || path.length() == 0) {
+ throw new TableException("Path should not be null or empty: " + path);
+ }
+
+ StringBuilder sb = new StringBuilder();
+ for (int i = 0; i < path.length(); i++) {
+ char c = path.charAt(i);
+ if (needsEscaping(c)) {
+ sb.append('%');
+ sb.append(String.format("%1$02X", (int) c));
+ } else {
+ sb.append(c);
+ }
+ }
+ return sb.toString();
+ }
+
+ /**
+ * Generates partition values from path.
+ *
+ * @param currPath Partition file path
+ * @param hivePartition Whether the partition path is with Hive style
+ * @param partitionKeys Partition keys
+ * @return Sequential partition specs.
+ */
+ public static List<String> extractPartitionValues(
+ Path currPath,
+ boolean hivePartition,
+ String[] partitionKeys) {
+ return new ArrayList<>(extractPartitionKeyValues(currPath, hivePartition, partitionKeys).values());
+ }
+
+ /**
+ * Generates partition key value mapping from path.
+ *
+ * @param currPath Partition file path
+ * @param hivePartition Whether the partition path is with Hive style
+ * @param partitionKeys Partition keys
+ * @return Sequential partition specs.
+ */
+ public static LinkedHashMap<String, String> extractPartitionKeyValues(
+ Path currPath,
+ boolean hivePartition,
+ String[] partitionKeys) {
+ LinkedHashMap<String, String> fullPartSpec = new LinkedHashMap<>();
+ List<String[]> kvs = new ArrayList<>();
+ int curDepth = 0;
+ do {
+ String component = currPath.getName();
+ final String[] kv = new String[2];
+ if (hivePartition) {
+ Matcher m = HIVE_PARTITION_NAME_PATTERN.matcher(component);
+ if (m.matches()) {
+ String k = unescapePathName(m.group(1));
+ String v = unescapePathName(m.group(2));
+ kv[0] = k;
+ kv[1] = v;
+ }
+ } else {
+ kv[0] = partitionKeys[partitionKeys.length - 1 - curDepth];
+ kv[1] = unescapePathName(component);
+ }
+ kvs.add(kv);
+ currPath = currPath.getParent();
+ curDepth++;
+ } while (currPath != null && !currPath.getName().isEmpty() && curDepth < partitionKeys.length);
+
+ // reverse the list since we checked the part from leaf dir to table's base dir
+ for (int i = kvs.size(); i > 0; i--) {
+ fullPartSpec.put(kvs.get(i - 1)[0], kvs.get(i - 1)[1]);
+ }
+
+ return fullPartSpec;
+ }
+
+ public static String unescapePathName(String path) {
+ StringBuilder sb = new StringBuilder();
+ for (int i = 0; i < path.length(); i++) {
+ char c = path.charAt(i);
+ if (c == '%' && i + 2 < path.length()) {
+ int code = -1;
+ try {
+ code = Integer.parseInt(path.substring(i + 1, i + 3), 16);
+ } catch (Exception ignored) {
+ // do nothing
+ }
+ if (code >= 0) {
+ sb.append((char) code);
+ i += 2;
+ continue;
+ }
+ }
+ sb.append(c);
+ }
+ return sb.toString();
+ }
+
+ /**
+ * Search all partitions in this path.
+ *
+ * @param fs File system
+ * @param path Search path
+ * @param hivePartition Whether the partition path is with Hive style
+ * @param partitionKeys Partition keys
+ * @return all partition key value mapping in sequence of the given path
+ */
+ public static List<Tuple2<LinkedHashMap<String, String>, Path>> searchPartKeyValueAndPaths(
+ FileSystem fs,
+ Path path,
+ boolean hivePartition,
+ String[] partitionKeys) {
+ // expectLevel start from 0, E.G. base_path/level0/level1/level2
+ FileStatus[] generatedParts = getFileStatusRecursively(path, partitionKeys.length, fs);
+ List<Tuple2<LinkedHashMap<String, String>, Path>> ret = new ArrayList<>();
+ for (FileStatus part : generatedParts) {
+ ret.add(
+ new Tuple2<>(
+ extractPartitionKeyValues(part.getPath(), hivePartition, partitionKeys),
+ part.getPath()));
+ }
+ return ret;
+ }
+
+ private static FileStatus[] getFileStatusRecursively(Path path, int expectLevel, FileSystem fs) {
+ ArrayList<FileStatus> result = new ArrayList<>();
+
+ try {
+ FileStatus fileStatus = fs.getFileStatus(path);
+ listStatusRecursively(fs, fileStatus, 0, expectLevel, result);
+ } catch (IOException ignore) {
+ return new FileStatus[0];
+ }
+
+ return result.toArray(new FileStatus[0]);
+ }
+
+ private static void listStatusRecursively(
+ FileSystem fs,
+ FileStatus fileStatus,
+ int level,
+ int expectLevel,
+ List<FileStatus> results) throws IOException {
+ if (expectLevel == level && !isHiddenFile(fileStatus)) {
+ results.add(fileStatus);
+ return;
+ }
+
+ if (fileStatus.isDir() && !isHiddenFile(fileStatus)) {
+ for (FileStatus stat : fs.listStatus(fileStatus.getPath())) {
+ listStatusRecursively(fs, stat, level + 1, expectLevel, results);
+ }
+ }
+ }
+
+ private static boolean isHiddenFile(FileStatus fileStatus) {
+ String name = fileStatus.getPath().getName();
+ return name.startsWith("_") || name.startsWith(".");
+ }
+
+ /**
+ * Same as getFileStatusRecursively but returns hadoop {@link org.apache.hadoop.fs.FileStatus}s.
Review comment:
return hadoop FileStatus?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] danny0405 commented on a change in pull request #2613: [HUDI-1647] Supports snapshot read for Flink
Posted by GitBox <gi...@apache.org>.
danny0405 commented on a change in pull request #2613:
URL: https://github.com/apache/hudi/pull/2613#discussion_r586099492
##########
File path: hudi-flink/src/main/java/org/apache/hudi/source/format/FilePathUtils.java
##########
@@ -0,0 +1,320 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.source.format;
+
+import org.apache.hudi.common.fs.FSUtils;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.core.fs.FileStatus;
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.table.api.TableException;
+import org.apache.hadoop.conf.Configuration;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.BitSet;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+/**
+ * Reference the Flink {@link org.apache.flink.table.utils.PartitionPathUtils}
+ * but supports simple partition path besides the Hive style.
+ */
+public class FilePathUtils {
+
+ private static final Pattern HIVE_PARTITION_NAME_PATTERN = Pattern.compile("([^/]+)=([^/]+)");
+
+ private static final BitSet CHAR_TO_ESCAPE = new BitSet(128);
+
+ static {
+ for (char c = 0; c < ' '; c++) {
+ CHAR_TO_ESCAPE.set(c);
+ }
+
+ /*
+ * ASCII 01-1F are HTTP control characters that need to be escaped.
+ * \u000A and \u000D are \n and \r, respectively.
+ */
+ char[] clist = new char[] {'\u0001', '\u0002', '\u0003', '\u0004',
+ '\u0005', '\u0006', '\u0007', '\u0008', '\u0009', '\n', '\u000B',
+ '\u000C', '\r', '\u000E', '\u000F', '\u0010', '\u0011', '\u0012',
+ '\u0013', '\u0014', '\u0015', '\u0016', '\u0017', '\u0018', '\u0019',
+ '\u001A', '\u001B', '\u001C', '\u001D', '\u001E', '\u001F',
+ '"', '#', '%', '\'', '*', '/', ':', '=', '?', '\\', '\u007F', '{',
+ '[', ']', '^'};
+
+ for (char c : clist) {
+ CHAR_TO_ESCAPE.set(c);
+ }
+ }
+
+ private static boolean needsEscaping(char c) {
+ return c < CHAR_TO_ESCAPE.size() && CHAR_TO_ESCAPE.get(c);
+ }
+
+ /**
+ * Make partition path from partition spec.
+ *
+ * @param partitionKVs The partition key value mapping
+ * @param hiveStylePartition Whether the partition path is with Hive style,
+ * e.g. {partition key} = {partition value}
+ * @return an escaped, valid partition name
+ */
+ public static String generatePartitionPath(
+ LinkedHashMap<String, String> partitionKVs,
+ boolean hiveStylePartition) {
+ if (partitionKVs.isEmpty()) {
+ return "";
+ }
+ StringBuilder suffixBuf = new StringBuilder();
+ int i = 0;
+ for (Map.Entry<String, String> e : partitionKVs.entrySet()) {
+ if (i > 0) {
+ suffixBuf.append(Path.SEPARATOR);
+ }
+ if (hiveStylePartition) {
+ suffixBuf.append(escapePathName(e.getKey()));
+ suffixBuf.append('=');
+ }
+ suffixBuf.append(escapePathName(e.getValue()));
+ i++;
+ }
+ suffixBuf.append(Path.SEPARATOR);
+ return suffixBuf.toString();
+ }
+
+ /**
+ * Escapes a path name.
+ *
+ * @param path The path to escape.
+ * @return An escaped path name.
+ */
+ private static String escapePathName(String path) {
+ if (path == null || path.length() == 0) {
+ throw new TableException("Path should not be null or empty: " + path);
+ }
+
+ StringBuilder sb = new StringBuilder();
+ for (int i = 0; i < path.length(); i++) {
+ char c = path.charAt(i);
+ if (needsEscaping(c)) {
+ sb.append('%');
+ sb.append(String.format("%1$02X", (int) c));
+ } else {
+ sb.append(c);
+ }
+ }
+ return sb.toString();
+ }
+
+ /**
+ * Generates partition values from path.
+ *
+ * @param currPath Partition file path
+ * @param hivePartition Whether the partition path is with Hive style
+ * @param partitionKeys Partition keys
+ * @return Sequential partition specs.
+ */
+ public static List<String> extractPartitionValues(
+ Path currPath,
+ boolean hivePartition,
+ String[] partitionKeys) {
+ return new ArrayList<>(extractPartitionKeyValues(currPath, hivePartition, partitionKeys).values());
+ }
+
+ /**
+ * Generates partition key value mapping from path.
+ *
+ * @param currPath Partition file path
+ * @param hivePartition Whether the partition path is with Hive style
+ * @param partitionKeys Partition keys
+ * @return Sequential partition specs.
+ */
+ public static LinkedHashMap<String, String> extractPartitionKeyValues(
+ Path currPath,
+ boolean hivePartition,
+ String[] partitionKeys) {
+ LinkedHashMap<String, String> fullPartSpec = new LinkedHashMap<>();
+ List<String[]> kvs = new ArrayList<>();
+ int curDepth = 0;
+ do {
+ String component = currPath.getName();
+ final String[] kv = new String[2];
+ if (hivePartition) {
+ Matcher m = HIVE_PARTITION_NAME_PATTERN.matcher(component);
+ if (m.matches()) {
+ String k = unescapePathName(m.group(1));
+ String v = unescapePathName(m.group(2));
+ kv[0] = k;
+ kv[1] = v;
+ }
+ } else {
+ kv[0] = partitionKeys[partitionKeys.length - 1 - curDepth];
+ kv[1] = unescapePathName(component);
+ }
+ kvs.add(kv);
+ currPath = currPath.getParent();
+ curDepth++;
+ } while (currPath != null && !currPath.getName().isEmpty() && curDepth < partitionKeys.length);
+
+ // reverse the list since we checked the part from leaf dir to table's base dir
+ for (int i = kvs.size(); i > 0; i--) {
+ fullPartSpec.put(kvs.get(i - 1)[0], kvs.get(i - 1)[1]);
+ }
+
+ return fullPartSpec;
+ }
+
+ public static String unescapePathName(String path) {
+ StringBuilder sb = new StringBuilder();
+ for (int i = 0; i < path.length(); i++) {
+ char c = path.charAt(i);
+ if (c == '%' && i + 2 < path.length()) {
+ int code = -1;
+ try {
+ code = Integer.parseInt(path.substring(i + 1, i + 3), 16);
+ } catch (Exception ignored) {
+ // do nothing
+ }
+ if (code >= 0) {
+ sb.append((char) code);
+ i += 2;
+ continue;
+ }
+ }
+ sb.append(c);
+ }
+ return sb.toString();
+ }
+
+ /**
+ * Search all partitions in this path.
+ *
+ * @param fs File system
+ * @param path Search path
+ * @param hivePartition Whether the partition path is with Hive style
+ * @param partitionKeys Partition keys
+ * @return all partition key value mapping in sequence of the given path
+ */
+ public static List<Tuple2<LinkedHashMap<String, String>, Path>> searchPartKeyValueAndPaths(
+ FileSystem fs,
+ Path path,
+ boolean hivePartition,
+ String[] partitionKeys) {
+ // expectLevel start from 0, E.G. base_path/level0/level1/level2
+ FileStatus[] generatedParts = getFileStatusRecursively(path, partitionKeys.length, fs);
+ List<Tuple2<LinkedHashMap<String, String>, Path>> ret = new ArrayList<>();
+ for (FileStatus part : generatedParts) {
+ ret.add(
+ new Tuple2<>(
+ extractPartitionKeyValues(part.getPath(), hivePartition, partitionKeys),
+ part.getPath()));
+ }
+ return ret;
+ }
+
+ private static FileStatus[] getFileStatusRecursively(Path path, int expectLevel, FileSystem fs) {
+ ArrayList<FileStatus> result = new ArrayList<>();
+
+ try {
+ FileStatus fileStatus = fs.getFileStatus(path);
+ listStatusRecursively(fs, fileStatus, 0, expectLevel, result);
+ } catch (IOException ignore) {
+ return new FileStatus[0];
+ }
+
+ return result.toArray(new FileStatus[0]);
+ }
+
+ private static void listStatusRecursively(
+ FileSystem fs,
+ FileStatus fileStatus,
+ int level,
+ int expectLevel,
+ List<FileStatus> results) throws IOException {
+ if (expectLevel == level && !isHiddenFile(fileStatus)) {
+ results.add(fileStatus);
+ return;
+ }
+
+ if (fileStatus.isDir() && !isHiddenFile(fileStatus)) {
+ for (FileStatus stat : fs.listStatus(fileStatus.getPath())) {
+ listStatusRecursively(fs, stat, level + 1, expectLevel, results);
+ }
+ }
+ }
+
+ private static boolean isHiddenFile(FileStatus fileStatus) {
+ String name = fileStatus.getPath().getName();
+ return name.startsWith("_") || name.startsWith(".");
+ }
+
+ /**
+ * Same as getFileStatusRecursively but returns hadoop {@link org.apache.hadoop.fs.FileStatus}s.
Review comment:
Yes.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] codecov-io edited a comment on pull request #2613: [HUDI-1647] Supports snapshot read for Flink
Posted by GitBox <gi...@apache.org>.
codecov-io edited a comment on pull request #2613:
URL: https://github.com/apache/hudi/pull/2613#issuecomment-787825622
# [Codecov](https://codecov.io/gh/apache/hudi/pull/2613?src=pr&el=h1) Report
> Merging [#2613](https://codecov.io/gh/apache/hudi/pull/2613?src=pr&el=desc) (0d77db7) into [master](https://codecov.io/gh/apache/hudi/commit/7a11de12764d8f68f296c6e68a22822318bfbefa?el=desc) (7a11de1) will **decrease** coverage by `0.34%`.
> The diff coverage is `47.23%`.
[](https://codecov.io/gh/apache/hudi/pull/2613?src=pr&el=tree)
```diff
@@ Coverage Diff @@
## master #2613 +/- ##
============================================
- Coverage 51.56% 51.22% -0.35%
- Complexity 3286 3292 +6
============================================
Files 445 440 -5
Lines 20328 20231 -97
Branches 2102 2161 +59
============================================
- Hits 10483 10363 -120
- Misses 8978 8980 +2
- Partials 867 888 +21
```
| Flag | Coverage Δ | Complexity Δ | |
|---|---|---|---|
| hudicli | `36.87% <ø> (ø)` | `0.00 <ø> (ø)` | |
| hudiclient | `100.00% <ø> (ø)` | `0.00 <ø> (ø)` | |
| hudicommon | `51.32% <ø> (-0.01%)` | `0.00 <ø> (ø)` | |
| hudiflink | `50.31% <47.23%> (-1.09%)` | `0.00 <188.00> (ø)` | |
| hudihadoopmr | `33.16% <ø> (ø)` | `0.00 <ø> (ø)` | |
| hudisparkdatasource | `69.71% <ø> (ø)` | `0.00 <ø> (ø)` | |
| hudisync | `?` | `?` | |
| huditimelineservice | `?` | `?` | |
| hudiutilities | `69.59% <ø> (+0.05%)` | `0.00 <ø> (ø)` | |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags#carryforward-flags-in-the-pull-request-comment) to find out more.
| [Impacted Files](https://codecov.io/gh/apache/hudi/pull/2613?src=pr&el=tree) | Coverage Δ | Complexity Δ | |
|---|---|---|---|
| [...va/org/apache/hudi/factory/HoodieTableFactory.java](https://codecov.io/gh/apache/hudi/pull/2613/diff?src=pr&el=tree#diff-aHVkaS1mbGluay9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvaHVkaS9mYWN0b3J5L0hvb2RpZVRhYmxlRmFjdG9yeS5qYXZh) | `0.00% <0.00%> (ø)` | `0.00 <0.00> (?)` | |
| [.../org/apache/hudi/operator/StreamWriteFunction.java](https://codecov.io/gh/apache/hudi/pull/2613/diff?src=pr&el=tree#diff-aHVkaS1mbGluay9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvaHVkaS9vcGVyYXRvci9TdHJlYW1Xcml0ZUZ1bmN0aW9uLmphdmE=) | `84.00% <0.00%> (-2.60%)` | `22.00 <0.00> (ø)` | |
| [.../org/apache/hudi/operator/StreamWriteOperator.java](https://codecov.io/gh/apache/hudi/pull/2613/diff?src=pr&el=tree#diff-aHVkaS1mbGluay9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvaHVkaS9vcGVyYXRvci9TdHJlYW1Xcml0ZU9wZXJhdG9yLmphdmE=) | `0.00% <0.00%> (ø)` | `0.00 <0.00> (ø)` | |
| [...ache/hudi/operator/StreamWriteOperatorFactory.java](https://codecov.io/gh/apache/hudi/pull/2613/diff?src=pr&el=tree#diff-aHVkaS1mbGluay9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvaHVkaS9vcGVyYXRvci9TdHJlYW1Xcml0ZU9wZXJhdG9yRmFjdG9yeS5qYXZh) | `0.00% <0.00%> (ø)` | `0.00 <0.00> (ø)` | |
| [...udi/operator/partitioner/BucketAssignFunction.java](https://codecov.io/gh/apache/hudi/pull/2613/diff?src=pr&el=tree#diff-aHVkaS1mbGluay9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvaHVkaS9vcGVyYXRvci9wYXJ0aXRpb25lci9CdWNrZXRBc3NpZ25GdW5jdGlvbi5qYXZh) | `85.86% <ø> (+0.92%)` | `22.00 <0.00> (+1.00)` | |
| [...ain/java/org/apache/hudi/sink/HoodieTableSink.java](https://codecov.io/gh/apache/hudi/pull/2613/diff?src=pr&el=tree#diff-aHVkaS1mbGluay9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvaHVkaS9zaW5rL0hvb2RpZVRhYmxlU2luay5qYXZh) | `0.00% <0.00%> (ø)` | `0.00 <0.00> (?)` | |
| [...e/hudi/source/format/cow/ParquetDecimalVector.java](https://codecov.io/gh/apache/hudi/pull/2613/diff?src=pr&el=tree#diff-aHVkaS1mbGluay9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvaHVkaS9zb3VyY2UvZm9ybWF0L2Nvdy9QYXJxdWV0RGVjaW1hbFZlY3Rvci5qYXZh) | `0.00% <0.00%> (ø)` | `0.00 <0.00> (?)` | |
| [.../org/apache/hudi/util/AvroToRowDataConverters.java](https://codecov.io/gh/apache/hudi/pull/2613/diff?src=pr&el=tree#diff-aHVkaS1mbGluay9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvaHVkaS91dGlsL0F2cm9Ub1Jvd0RhdGFDb252ZXJ0ZXJzLmphdmE=) | `16.94% <16.94%> (ø)` | `12.00 <12.00> (?)` | |
| [...hudi/source/format/cow/ParquetSplitReaderUtil.java](https://codecov.io/gh/apache/hudi/pull/2613/diff?src=pr&el=tree#diff-aHVkaS1mbGluay9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvaHVkaS9zb3VyY2UvZm9ybWF0L2Nvdy9QYXJxdWV0U3BsaXRSZWFkZXJVdGlsLmphdmE=) | `20.00% <20.00%> (ø)` | `13.00 <13.00> (?)` | |
| [...java/org/apache/hudi/util/AvroSchemaConverter.java](https://codecov.io/gh/apache/hudi/pull/2613/diff?src=pr&el=tree#diff-aHVkaS1mbGluay9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvaHVkaS91dGlsL0F2cm9TY2hlbWFDb252ZXJ0ZXIuamF2YQ==) | `23.74% <25.97%> (+23.74%)` | `13.00 <7.00> (+13.00)` | |
| ... and [60 more](https://codecov.io/gh/apache/hudi/pull/2613/diff?src=pr&el=tree-more) | |
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] danny0405 commented on a change in pull request #2613: [HUDI-1647] Supports snapshot read for Flink
Posted by GitBox <gi...@apache.org>.
danny0405 commented on a change in pull request #2613:
URL: https://github.com/apache/hudi/pull/2613#discussion_r586098383
##########
File path: hudi-flink/src/main/java/org/apache/hudi/source/format/cow/AbstractColumnReader.java
##########
@@ -0,0 +1,324 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.source.format.cow;
+
+import org.apache.flink.formats.parquet.vector.ParquetDictionary;
+import org.apache.flink.formats.parquet.vector.reader.ColumnReader;
+import org.apache.flink.table.data.vector.writable.WritableColumnVector;
+import org.apache.flink.table.data.vector.writable.WritableIntVector;
+import org.apache.parquet.Preconditions;
+import org.apache.parquet.bytes.ByteBufferInputStream;
+import org.apache.parquet.bytes.BytesInput;
+import org.apache.parquet.bytes.BytesUtils;
+import org.apache.parquet.column.ColumnDescriptor;
+import org.apache.parquet.column.Dictionary;
+import org.apache.parquet.column.Encoding;
+import org.apache.parquet.column.page.DataPage;
+import org.apache.parquet.column.page.DataPageV1;
+import org.apache.parquet.column.page.DataPageV2;
+import org.apache.parquet.column.page.DictionaryPage;
+import org.apache.parquet.column.page.PageReader;
+import org.apache.parquet.column.values.ValuesReader;
+import org.apache.parquet.io.ParquetDecodingException;
+import org.apache.parquet.schema.PrimitiveType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+
+import static org.apache.parquet.column.ValuesType.REPETITION_LEVEL;
+
+/**
+ * Abstract {@link ColumnReader}.
+ * See {@link org.apache.parquet.column.impl.ColumnReaderImpl},
+ * part of the code is referred from Apache Spark and Apache Parquet.
+ *
+ * <p>Note: Reference Flink release 1.11.2 {@link org.apache.flink.formats.parquet.vector.reader.AbstractColumnReader}
+ * because some of the package scope methods.
+ */
+public abstract class AbstractColumnReader<V extends WritableColumnVector>
+ implements ColumnReader<V> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(org.apache.flink.formats.parquet.vector.reader.AbstractColumnReader.class);
+
+ private final PageReader pageReader;
+
+ /**
+ * The dictionary, if this column has dictionary encoding.
+ */
+ protected final Dictionary dictionary;
+
+ /**
+ * Maximum definition level for this column.
+ */
+ protected final int maxDefLevel;
+
+ protected final ColumnDescriptor descriptor;
+
+ /**
+ * Total number of values read.
+ */
+ private long valuesRead;
+
+ /**
+ * value that indicates the end of the current page. That is, if valuesRead ==
+ * endOfPageValueCount, we are at the end of the page.
+ */
+ private long endOfPageValueCount;
+
+ /**
+ * If true, the current page is dictionary encoded.
+ */
+ private boolean isCurrentPageDictionaryEncoded;
+
+ /**
+ * Total values in the current page.
+ */
+ private int pageValueCount;
+
+ /*
+ * Input streams:
+ * 1.Run length encoder to encode every data, so we have run length stream to get
+ * run length information.
+ * 2.Data maybe is real data, maybe is dictionary ids which need be decode to real
+ * data from Dictionary.
+ *
+ * Run length stream ------> Data stream
+ * |
+ * ------> Dictionary ids stream
+ */
+
+ /**
+ * Run length decoder for data and dictionary.
+ */
+ protected RunLengthDecoder runLenDecoder;
+
+ /**
+ * Data input stream.
+ */
+ ByteBufferInputStream dataInputStream;
+
+ /**
+ * Dictionary decoder to wrap dictionary ids input stream.
+ */
+ private RunLengthDecoder dictionaryIdsDecoder;
+
+ public AbstractColumnReader(
+ ColumnDescriptor descriptor,
+ PageReader pageReader) throws IOException {
+ this.descriptor = descriptor;
+ this.pageReader = pageReader;
+ this.maxDefLevel = descriptor.getMaxDefinitionLevel();
+
+ DictionaryPage dictionaryPage = pageReader.readDictionaryPage();
+ if (dictionaryPage != null) {
+ try {
+ this.dictionary = dictionaryPage.getEncoding().initDictionary(descriptor, dictionaryPage);
+ this.isCurrentPageDictionaryEncoded = true;
+ } catch (IOException e) {
+ throw new IOException("could not decode the dictionary for " + descriptor, e);
+ }
+ } else {
+ this.dictionary = null;
+ this.isCurrentPageDictionaryEncoded = false;
+ }
+ /*
+ * Total number of values in this column (in this row group).
+ */
+ long totalValueCount = pageReader.getTotalValueCount();
+ if (totalValueCount == 0) {
+ throw new IOException("totalValueCount == 0");
+ }
+ }
+
+ protected void checkTypeName(PrimitiveType.PrimitiveTypeName expectedName) {
+ PrimitiveType.PrimitiveTypeName actualName = descriptor.getPrimitiveType().getPrimitiveTypeName();
+ Preconditions.checkArgument(
+ actualName == expectedName,
+ "Expected type name: %s, actual type name: %s",
+ expectedName,
+ actualName);
+ }
+
+ /**
+ * Reads `total` values from this columnReader into column.
+ */
+ @Override
+ public final void readToVector(int readNumber, V vector) throws IOException {
+ int rowId = 0;
+ WritableIntVector dictionaryIds = null;
+ if (dictionary != null) {
+ dictionaryIds = vector.reserveDictionaryIds(readNumber);
+ }
+ while (readNumber > 0) {
+ // Compute the number of values we want to read in this page.
+ int leftInPage = (int) (endOfPageValueCount - valuesRead);
+ if (leftInPage == 0) {
+ DataPage page = pageReader.readPage();
+ if (page instanceof DataPageV1) {
+ readPageV1((DataPageV1) page);
+ } else if (page instanceof DataPageV2) {
+ readPageV2((DataPageV2) page);
+ } else {
+ throw new RuntimeException("Unsupported page type: " + page.getClass());
+ }
+ leftInPage = (int) (endOfPageValueCount - valuesRead);
+ }
+ int num = Math.min(readNumber, leftInPage);
+ if (isCurrentPageDictionaryEncoded) {
+ // Read and decode dictionary ids.
+ runLenDecoder.readDictionaryIds(
+ num, dictionaryIds, vector, rowId, maxDefLevel, this.dictionaryIdsDecoder);
+
+ if (vector.hasDictionary() || (rowId == 0 && supportLazyDecode())) {
+ // Column vector supports lazy decoding of dictionary values so just set the dictionary.
+ // We can't do this if rowId != 0 AND the column doesn't have a dictionary (i.e. some
+ // non-dictionary encoded values have already been added).
+ vector.setDictionary(new ParquetDictionary(dictionary));
+ } else {
+ readBatchFromDictionaryIds(rowId, num, vector, dictionaryIds);
+ }
+ } else {
+ if (vector.hasDictionary() && rowId != 0) {
+ // This batch already has dictionary encoded values but this new page is not. The batch
+ // does not support a mix of dictionary and not so we will decode the dictionary.
+ readBatchFromDictionaryIds(0, rowId, vector, vector.getDictionaryIds());
+ }
+ vector.setDictionary(null);
+ readBatch(rowId, num, vector);
+ }
+
+ valuesRead += num;
+ rowId += num;
+ readNumber -= num;
+ }
+ }
+
+ private void readPageV1(DataPageV1 page) throws IOException {
+ this.pageValueCount = page.getValueCount();
+ ValuesReader rlReader = page.getRlEncoding().getValuesReader(descriptor, REPETITION_LEVEL);
+
+ // Initialize the decoders.
+ if (page.getDlEncoding() != Encoding.RLE && descriptor.getMaxDefinitionLevel() != 0) {
+ throw new UnsupportedOperationException("Unsupported encoding: " + page.getDlEncoding());
+ }
+ int bitWidth = BytesUtils.getWidthFromMaxInt(descriptor.getMaxDefinitionLevel());
+ this.runLenDecoder = new RunLengthDecoder(bitWidth);
+ try {
+ BytesInput bytes = page.getBytes();
+ ByteBufferInputStream in = bytes.toInputStream();
+ rlReader.initFromPage(pageValueCount, in);
+ this.runLenDecoder.initFromStream(pageValueCount, in);
+ prepareNewPage(page.getValueEncoding(), in);
+ } catch (IOException e) {
+ throw new IOException("could not read page " + page + " in col " + descriptor, e);
+ }
+ }
+
+ private void readPageV2(DataPageV2 page) throws IOException {
+ this.pageValueCount = page.getValueCount();
+
+ int bitWidth = BytesUtils.getWidthFromMaxInt(descriptor.getMaxDefinitionLevel());
+ // do not read the length from the stream. v2 pages handle dividing the page bytes.
+ this.runLenDecoder = new RunLengthDecoder(bitWidth, false);
+ this.runLenDecoder.initFromStream(
+ this.pageValueCount, page.getDefinitionLevels().toInputStream());
+ try {
+ prepareNewPage(page.getDataEncoding(), page.getData().toInputStream());
+ } catch (IOException e) {
+ throw new IOException("could not read page " + page + " in col " + descriptor, e);
+ }
+ }
+
+ private void prepareNewPage(
+ Encoding dataEncoding,
+ ByteBufferInputStream in) throws IOException {
+ this.endOfPageValueCount = valuesRead + pageValueCount;
+ if (dataEncoding.usesDictionary()) {
+ if (dictionary == null) {
+ throw new IOException("Could not read page in col "
+ + descriptor
+ + " as the dictionary was missing for encoding "
+ + dataEncoding);
+ }
+ @SuppressWarnings("deprecation")
+ Encoding plainDict = Encoding.PLAIN_DICTIONARY; // var to allow warning suppression
+ if (dataEncoding != plainDict && dataEncoding != Encoding.RLE_DICTIONARY) {
+ throw new UnsupportedOperationException("Unsupported encoding: " + dataEncoding);
+ }
+ this.dataInputStream = null;
+ this.dictionaryIdsDecoder = new RunLengthDecoder();
+ try {
+ this.dictionaryIdsDecoder.initFromStream(pageValueCount, in);
+ } catch (IOException e) {
+ throw new IOException("could not read dictionary in col " + descriptor, e);
+ }
+ this.isCurrentPageDictionaryEncoded = true;
+ } else {
+ if (dataEncoding != Encoding.PLAIN) {
+ throw new UnsupportedOperationException("Unsupported encoding: " + dataEncoding);
+ }
+ this.dictionaryIdsDecoder = null;
+ LOG.debug("init from page at offset {} for length {}", in.position(), in.available());
+ this.dataInputStream = in.remainingStream();
+ this.isCurrentPageDictionaryEncoded = false;
+ }
+
+ afterReadPage();
+ }
+
+ final ByteBuffer readDataBuffer(int length) {
+ try {
+ return dataInputStream.slice(length).order(ByteOrder.LITTLE_ENDIAN);
+ } catch (IOException e) {
+ throw new ParquetDecodingException("Failed to read " + length + " bytes", e);
+ }
+ }
+
+ /**
+ * After read a page, we may need some initialization.
+ */
+ protected void afterReadPage() {}
+
+ /**
+ * Support lazy dictionary ids decode. See more in {@link ParquetDictionary}.
+ * If return false, we will decode all the data first.
+ */
+ protected boolean supportLazyDecode() {
+ return true;
+ }
+
+ /**
+ * Read batch from {@link #runLenDecoder} and {@link #dataInputStream}.
+ */
+ protected abstract void readBatch(int rowId, int num, V column);
+
+ /**
+ * Decode dictionary ids to data.
+ * From {@link #runLenDecoder} and {@link #dictionaryIdsDecoder}.
+ */
+ protected abstract void readBatchFromDictionaryIds(
+ int rowId,
Review comment:
I think it's fine and more easy for future upgrade.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] danny0405 commented on a change in pull request #2613: [HUDI-1647] Supports snapshot read for Flink
Posted by GitBox <gi...@apache.org>.
danny0405 commented on a change in pull request #2613:
URL: https://github.com/apache/hudi/pull/2613#discussion_r586099356
##########
File path: hudi-flink/src/main/java/org/apache/hudi/source/HoodieTableSource.java
##########
@@ -0,0 +1,411 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.source;
+
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.model.HoodieBaseFile;
+import org.apache.hudi.common.model.HoodieLogFile;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.TableSchemaResolver;
+import org.apache.hudi.common.table.view.HoodieTableFileSystemView;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.hadoop.HoodieROTablePathFilter;
+import org.apache.hudi.hadoop.utils.HoodieRealtimeInputFormatUtils;
+import org.apache.hudi.operator.FlinkOptions;
+import org.apache.hudi.source.format.FilePathUtils;
+import org.apache.hudi.source.format.cow.CopyOnWriteInputFormat;
+import org.apache.hudi.source.format.mor.MergeOnReadInputFormat;
+import org.apache.hudi.source.format.mor.MergeOnReadInputSplit;
+import org.apache.hudi.source.format.mor.MergeOnReadTableState;
+import org.apache.hudi.util.AvroSchemaConverter;
+import org.apache.hudi.util.StreamerUtil;
+
+import org.apache.avro.Schema;
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.common.io.FileInputFormat;
+import org.apache.flink.api.common.io.FilePathFilter;
+import org.apache.flink.api.common.io.InputFormat;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.io.CollectionInputFormat;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.DataStreamSource;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.source.InputFormatSourceFunction;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.expressions.Expression;
+import org.apache.flink.table.runtime.types.TypeInfoDataTypeConverter;
+import org.apache.flink.table.sources.FilterableTableSource;
+import org.apache.flink.table.sources.LimitableTableSource;
+import org.apache.flink.table.sources.PartitionableTableSource;
+import org.apache.flink.table.sources.ProjectableTableSource;
+import org.apache.flink.table.sources.StreamTableSource;
+import org.apache.flink.table.sources.TableSource;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.mapred.JobConf;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import static org.apache.hudi.hadoop.utils.HoodieRealtimeRecordReaderUtils.getMaxCompactionMemoryInBytes;
+import static org.apache.hudi.source.format.FormatUtils.getParquetConf;
+
+/**
+ * Hoodie batch table source that always read the latest snapshot of the underneath table.
+ */
+public class HoodieTableSource implements
+ StreamTableSource<RowData>,
+ PartitionableTableSource,
+ ProjectableTableSource<RowData>,
+ LimitableTableSource<RowData>,
+ FilterableTableSource<RowData> {
+ private static final Logger LOG = LoggerFactory.getLogger(HoodieTableSource.class);
+
+ private static final int NO_LIMIT_CONSTANT = -1;
+
+ private final transient org.apache.hadoop.conf.Configuration hadoopConf;
+ private final transient HoodieTableMetaClient metaClient;
+ private final long maxCompactionMemoryInBytes;
+
+ private final TableSchema schema;
+ private final Path path;
+ private final List<String> partitionKeys;
+ private final String defaultPartName;
+ private final Configuration conf;
+
+ private final int[] requiredPos;
+ private final long limit;
+ private final List<Expression> filters;
+
+ private List<Map<String, String>> requiredPartitions;
+
+ public HoodieTableSource(
+ TableSchema schema,
+ Path path,
+ List<String> partitionKeys,
+ String defaultPartName,
+ Configuration conf) {
+ this(schema, path, partitionKeys, defaultPartName, conf, null, null, null, null);
+ }
+
+ public HoodieTableSource(
+ TableSchema schema,
+ Path path,
+ List<String> partitionKeys,
+ String defaultPartName,
+ Configuration conf,
+ @Nullable List<Map<String, String>> requiredPartitions,
+ @Nullable int[] requiredPos,
+ @Nullable Long limit,
+ @Nullable List<Expression> filters) {
+ this.schema = schema;
+ this.path = path;
+ this.partitionKeys = partitionKeys;
+ this.defaultPartName = defaultPartName;
+ this.conf = conf;
+ this.requiredPartitions = requiredPartitions;
+ this.requiredPos = requiredPos == null
+ ? IntStream.range(0, schema.getFieldCount()).toArray()
+ : requiredPos;
+ this.limit = limit == null ? NO_LIMIT_CONSTANT : limit;
+ this.filters = filters == null ? Collections.emptyList() : filters;
+ final String basePath = this.conf.getString(FlinkOptions.PATH);
+ this.hadoopConf = StreamerUtil.getHadoopConf();
+ this.metaClient = HoodieTableMetaClient.builder().setConf(hadoopConf).setBasePath(basePath).build();
+ this.maxCompactionMemoryInBytes = getMaxCompactionMemoryInBytes(new JobConf(this.hadoopConf));
+ }
+
+ @Override
+ public DataStream<RowData> getDataStream(StreamExecutionEnvironment execEnv) {
+ @SuppressWarnings("unchecked")
+ TypeInformation<RowData> typeInfo =
+ (TypeInformation<RowData>) TypeInfoDataTypeConverter.fromDataTypeToTypeInfo(getProducedDataType());
+ InputFormatSourceFunction<RowData> func = new InputFormatSourceFunction<>(getInputFormat(), typeInfo);
+ DataStreamSource<RowData> source = execEnv.addSource(func, explainSource(), typeInfo);
+ return source.name(explainSource());
+ }
+
+ @Override
+ public boolean isBounded() {
+ return true;
+ }
+
+ @Override
+ public TableSource<RowData> applyPredicate(List<Expression> predicates) {
+ return new HoodieTableSource(schema, path, partitionKeys, defaultPartName, conf,
+ requiredPartitions, requiredPos, limit, new ArrayList<>(predicates));
+ }
+
+ @Override
+ public boolean isFilterPushedDown() {
+ return this.filters != null && this.filters.size() > 0;
+ }
+
+ @Override
+ public boolean isLimitPushedDown() {
+ return this.limit != NO_LIMIT_CONSTANT;
+ }
+
+ @Override
+ public TableSource<RowData> applyLimit(long limit) {
+ return new HoodieTableSource(schema, path, partitionKeys, defaultPartName, conf,
+ requiredPartitions, requiredPos, limit, filters);
+ }
+
+ @Override
+ public List<Map<String, String>> getPartitions() {
+ try {
+ return FilePathUtils
+ .searchPartKeyValueAndPaths(
+ path.getFileSystem(),
+ path,
+ conf.getBoolean(FlinkOptions.HIVE_STYLE_PARTITION),
+ partitionKeys.toArray(new String[0]))
+ .stream()
+ .map(tuple2 -> tuple2.f0)
+ .map(spec -> {
+ LinkedHashMap<String, String> ret = new LinkedHashMap<>();
+ spec.forEach((k, v) -> ret.put(k, defaultPartName.equals(v) ? null : v));
+ return ret;
+ })
+ .collect(Collectors.toList());
+ } catch (Exception e) {
+ throw new TableException("Fetch partitions fail.", e);
+ }
+ }
+
+ @Override
+ public TableSource applyPartitionPruning(List<Map<String, String>> requiredPartitions) {
+ return new HoodieTableSource(schema, path, partitionKeys, defaultPartName, conf,
+ requiredPartitions, requiredPos, limit, filters);
+ }
+
+ @Override
+ public TableSource<RowData> projectFields(int[] requiredPos) {
+ return new HoodieTableSource(schema, path, partitionKeys, defaultPartName, conf,
+ requiredPartitions, requiredPos, limit, filters);
+ }
+
+ @Override
+ public TableSchema getTableSchema() {
+ return schema;
+ }
+
+ @Override
+ public DataType getProducedDataType() {
+ String[] schemaFieldNames = this.schema.getFieldNames();
+ DataType[] schemaTypes = this.schema.getFieldDataTypes();
+
+ return DataTypes.ROW(Arrays.stream(this.requiredPos)
+ .mapToObj(i -> DataTypes.FIELD(schemaFieldNames[i], schemaTypes[i]))
+ .toArray(DataTypes.Field[]::new))
+ .bridgedTo(RowData.class);
+ }
+
+ private List<Map<String, String>> getOrFetchPartitions() {
+ if (requiredPartitions == null) {
+ requiredPartitions = getPartitions();
+ }
+ return requiredPartitions;
+ }
+
+ private List<MergeOnReadInputSplit> buildFileIndex(Path[] paths) {
+ FileStatus[] fileStatuses = Arrays.stream(paths)
+ .flatMap(path -> Arrays.stream(FilePathUtils.getFileStatusRecursivelyV2(path, 1, hadoopConf)))
+ .toArray(FileStatus[]::new);
+ if (fileStatuses.length == 0) {
+ throw new HoodieException("No files found for reading in user provided path.");
+ }
+
+ HoodieTableFileSystemView fsView = new HoodieTableFileSystemView(metaClient,
+ metaClient.getActiveTimeline().getCommitsTimeline()
+ .filterCompletedInstants(), fileStatuses);
+ List<HoodieBaseFile> latestFiles = fsView.getLatestBaseFiles().collect(Collectors.toList());
+ String latestCommit = fsView.getLastInstant().get().getTimestamp();
+ final String mergeType = this.conf.getString(FlinkOptions.MERGE_TYPE);
+ final AtomicInteger cnt = new AtomicInteger(0);
+ if (latestFiles.size() > 0) {
+ Map<HoodieBaseFile, List<String>> fileGroup =
+ HoodieRealtimeInputFormatUtils.groupLogsByBaseFile(hadoopConf, latestFiles);
+ return fileGroup.entrySet().stream().map(kv -> {
+ HoodieBaseFile baseFile = kv.getKey();
+ Option<List<String>> logPaths = kv.getValue().size() == 0
+ ? Option.empty()
+ : Option.of(kv.getValue());
+ return new MergeOnReadInputSplit(cnt.getAndAdd(1),
+ baseFile.getPath(), logPaths, latestCommit,
+ metaClient.getBasePath(), maxCompactionMemoryInBytes, mergeType);
+ }).collect(Collectors.toList());
+ } else {
+ // all the files are logs
Review comment:
Flink HUDI can index log files (say it's global index), thus, the Flink writer can write all log files for MOR table.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] danny0405 commented on a change in pull request #2613: [HUDI-1647] Supports snapshot read for Flink
Posted by GitBox <gi...@apache.org>.
danny0405 commented on a change in pull request #2613:
URL: https://github.com/apache/hudi/pull/2613#discussion_r586100203
##########
File path: hudi-flink/src/main/java/org/apache/hudi/source/format/cow/CopyOnWriteInputFormat.java
##########
@@ -0,0 +1,134 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.source.format.cow;
+
+import org.apache.flink.api.common.io.FileInputFormat;
+import org.apache.flink.core.fs.FileInputSplit;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.formats.parquet.utils.SerializableConfiguration;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.utils.PartitionPathUtils;
+import org.apache.hadoop.conf.Configuration;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.LinkedHashMap;
+import java.util.List;
+
+import static org.apache.flink.table.data.vector.VectorizedColumnBatch.DEFAULT_SIZE;
+import static org.apache.flink.table.filesystem.RowPartitionComputer.restorePartValueFromType;
+
+/**
+ * An implementation of {@link FileInputFormat} to read {@link RowData} records
+ * from Parquet files.
+ *
+ * <p>Note: Reference Flink release 1.11.2
+ * {@code org.apache.flink.formats.parquet.ParquetFileSystemFormatFactory.ParquetInputFormat}
+ * to support TIMESTAMP_MILLIS.
+ *
+ * @see ParquetSplitReaderUtil
+ */
+public class CopyOnWriteInputFormat extends FileInputFormat<RowData> {
Review comment:
In order to extend the TIMESTAMP INT 64 type which HUDI uses as a default, i have explained in the doc.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] codecov-io edited a comment on pull request #2613: [HUDI-1647] Supports snapshot read for Flink
Posted by GitBox <gi...@apache.org>.
codecov-io edited a comment on pull request #2613:
URL: https://github.com/apache/hudi/pull/2613#issuecomment-787825622
# [Codecov](https://codecov.io/gh/apache/hudi/pull/2613?src=pr&el=h1) Report
> Merging [#2613](https://codecov.io/gh/apache/hudi/pull/2613?src=pr&el=desc) (c0a3083) into [master](https://codecov.io/gh/apache/hudi/commit/7a11de12764d8f68f296c6e68a22822318bfbefa?el=desc) (7a11de1) will **decrease** coverage by `0.15%`.
> The diff coverage is `47.23%`.
[](https://codecov.io/gh/apache/hudi/pull/2613?src=pr&el=tree)
```diff
@@ Coverage Diff @@
## master #2613 +/- ##
============================================
- Coverage 51.56% 51.40% -0.16%
- Complexity 3286 3479 +193
============================================
Files 445 461 +16
Lines 20328 21675 +1347
Branches 2102 2299 +197
============================================
+ Hits 10483 11143 +660
- Misses 8978 9572 +594
- Partials 867 960 +93
```
| Flag | Coverage Δ | Complexity Δ | |
|---|---|---|---|
| hudicli | `36.87% <ø> (ø)` | `0.00 <ø> (ø)` | |
| hudiclient | `100.00% <ø> (ø)` | `0.00 <ø> (ø)` | |
| hudicommon | `51.30% <ø> (-0.03%)` | `0.00 <ø> (ø)` | |
| hudiflink | `50.36% <47.23%> (-1.03%)` | `0.00 <188.00> (ø)` | |
| hudihadoopmr | `33.16% <ø> (ø)` | `0.00 <ø> (ø)` | |
| hudisparkdatasource | `69.71% <ø> (ø)` | `0.00 <ø> (ø)` | |
| hudisync | `49.62% <ø> (ø)` | `0.00 <ø> (ø)` | |
| huditimelineservice | `66.49% <ø> (ø)` | `0.00 <ø> (ø)` | |
| hudiutilities | `69.54% <ø> (ø)` | `0.00 <ø> (ø)` | |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags#carryforward-flags-in-the-pull-request-comment) to find out more.
| [Impacted Files](https://codecov.io/gh/apache/hudi/pull/2613?src=pr&el=tree) | Coverage Δ | Complexity Δ | |
|---|---|---|---|
| [...va/org/apache/hudi/factory/HoodieTableFactory.java](https://codecov.io/gh/apache/hudi/pull/2613/diff?src=pr&el=tree#diff-aHVkaS1mbGluay9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvaHVkaS9mYWN0b3J5L0hvb2RpZVRhYmxlRmFjdG9yeS5qYXZh) | `0.00% <0.00%> (ø)` | `0.00 <0.00> (?)` | |
| [.../org/apache/hudi/operator/StreamWriteFunction.java](https://codecov.io/gh/apache/hudi/pull/2613/diff?src=pr&el=tree#diff-aHVkaS1mbGluay9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvaHVkaS9vcGVyYXRvci9TdHJlYW1Xcml0ZUZ1bmN0aW9uLmphdmE=) | `84.00% <0.00%> (-2.60%)` | `22.00 <0.00> (ø)` | |
| [.../org/apache/hudi/operator/StreamWriteOperator.java](https://codecov.io/gh/apache/hudi/pull/2613/diff?src=pr&el=tree#diff-aHVkaS1mbGluay9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvaHVkaS9vcGVyYXRvci9TdHJlYW1Xcml0ZU9wZXJhdG9yLmphdmE=) | `0.00% <0.00%> (ø)` | `0.00 <0.00> (ø)` | |
| [...ache/hudi/operator/StreamWriteOperatorFactory.java](https://codecov.io/gh/apache/hudi/pull/2613/diff?src=pr&el=tree#diff-aHVkaS1mbGluay9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvaHVkaS9vcGVyYXRvci9TdHJlYW1Xcml0ZU9wZXJhdG9yRmFjdG9yeS5qYXZh) | `0.00% <0.00%> (ø)` | `0.00 <0.00> (ø)` | |
| [...udi/operator/partitioner/BucketAssignFunction.java](https://codecov.io/gh/apache/hudi/pull/2613/diff?src=pr&el=tree#diff-aHVkaS1mbGluay9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvaHVkaS9vcGVyYXRvci9wYXJ0aXRpb25lci9CdWNrZXRBc3NpZ25GdW5jdGlvbi5qYXZh) | `85.86% <ø> (+0.92%)` | `22.00 <0.00> (+1.00)` | |
| [...ain/java/org/apache/hudi/sink/HoodieTableSink.java](https://codecov.io/gh/apache/hudi/pull/2613/diff?src=pr&el=tree#diff-aHVkaS1mbGluay9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvaHVkaS9zaW5rL0hvb2RpZVRhYmxlU2luay5qYXZh) | `0.00% <0.00%> (ø)` | `0.00 <0.00> (?)` | |
| [...e/hudi/source/format/cow/ParquetDecimalVector.java](https://codecov.io/gh/apache/hudi/pull/2613/diff?src=pr&el=tree#diff-aHVkaS1mbGluay9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvaHVkaS9zb3VyY2UvZm9ybWF0L2Nvdy9QYXJxdWV0RGVjaW1hbFZlY3Rvci5qYXZh) | `0.00% <0.00%> (ø)` | `0.00 <0.00> (?)` | |
| [.../org/apache/hudi/util/AvroToRowDataConverters.java](https://codecov.io/gh/apache/hudi/pull/2613/diff?src=pr&el=tree#diff-aHVkaS1mbGluay9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvaHVkaS91dGlsL0F2cm9Ub1Jvd0RhdGFDb252ZXJ0ZXJzLmphdmE=) | `16.94% <16.94%> (ø)` | `12.00 <12.00> (?)` | |
| [...hudi/source/format/cow/ParquetSplitReaderUtil.java](https://codecov.io/gh/apache/hudi/pull/2613/diff?src=pr&el=tree#diff-aHVkaS1mbGluay9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvaHVkaS9zb3VyY2UvZm9ybWF0L2Nvdy9QYXJxdWV0U3BsaXRSZWFkZXJVdGlsLmphdmE=) | `20.00% <20.00%> (ø)` | `13.00 <13.00> (?)` | |
| [...java/org/apache/hudi/util/AvroSchemaConverter.java](https://codecov.io/gh/apache/hudi/pull/2613/diff?src=pr&el=tree#diff-aHVkaS1mbGluay9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvaHVkaS91dGlsL0F2cm9TY2hlbWFDb252ZXJ0ZXIuamF2YQ==) | `23.74% <25.97%> (+23.74%)` | `13.00 <7.00> (+13.00)` | |
| ... and [36 more](https://codecov.io/gh/apache/hudi/pull/2613/diff?src=pr&el=tree-more) | |
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] danny0405 commented on a change in pull request #2613: [HUDI-1647] Supports snapshot read for Flink
Posted by GitBox <gi...@apache.org>.
danny0405 commented on a change in pull request #2613:
URL: https://github.com/apache/hudi/pull/2613#discussion_r586100912
##########
File path: hudi-flink/src/main/java/org/apache/hudi/source/format/cow/ParquetColumnarRowSplitReader.java
##########
@@ -0,0 +1,370 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.source.format.cow;
+
+import org.apache.flink.formats.parquet.vector.reader.ColumnReader;
+import org.apache.flink.table.data.ColumnarRowData;
+import org.apache.flink.table.data.vector.ColumnVector;
+import org.apache.flink.table.data.vector.VectorizedColumnBatch;
+import org.apache.flink.table.data.vector.writable.WritableColumnVector;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
+import org.apache.flink.util.FlinkRuntimeException;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.parquet.column.ColumnDescriptor;
+import org.apache.parquet.column.page.PageReadStore;
+import org.apache.parquet.filter2.compat.FilterCompat;
+import org.apache.parquet.hadoop.ParquetFileReader;
+import org.apache.parquet.hadoop.metadata.BlockMetaData;
+import org.apache.parquet.hadoop.metadata.ParquetMetadata;
+import org.apache.parquet.schema.GroupType;
+import org.apache.parquet.schema.MessageType;
+import org.apache.parquet.schema.Type;
+import org.apache.parquet.schema.Types;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+
+import static org.apache.hudi.source.format.cow.ParquetSplitReaderUtil.createColumnReader;
+import static org.apache.hudi.source.format.cow.ParquetSplitReaderUtil.createWritableColumnVector;
+import static org.apache.parquet.filter2.compat.RowGroupFilter.filterRowGroups;
+import static org.apache.parquet.format.converter.ParquetMetadataConverter.range;
+import static org.apache.parquet.hadoop.ParquetFileReader.readFooter;
+import static org.apache.parquet.hadoop.ParquetInputFormat.getFilter;
+
+/**
+ * This reader is used to read a {@link VectorizedColumnBatch} from input split.
+ *
+ * <p>Note: Reference Flink release 1.11.2
+ * {@code org.apache.flink.formats.parquet.vector.ParquetColumnarRowSplitReader}
+ * because it is package scope.
+ */
+public class ParquetColumnarRowSplitReader implements Closeable {
Review comment:
It is not that good to put the copied code, but it is still better than adding `flink-parquet` dependency and shade all the codes in the class path, which would cause conflicts and confuses the users (you can not prevent people to have the original `flink-parquet` jar in the classpath).
By copy several classes (basically unchanged overtime) we avoid that confusion.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] garyli1019 commented on a change in pull request #2613: [HUDI-1647] Supports snapshot read for Flink
Posted by GitBox <gi...@apache.org>.
garyli1019 commented on a change in pull request #2613:
URL: https://github.com/apache/hudi/pull/2613#discussion_r586180443
##########
File path: hudi-flink/src/main/java/org/apache/hudi/source/format/cow/ParquetColumnarRowSplitReader.java
##########
@@ -0,0 +1,370 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.source.format.cow;
+
+import org.apache.flink.formats.parquet.vector.reader.ColumnReader;
+import org.apache.flink.table.data.ColumnarRowData;
+import org.apache.flink.table.data.vector.ColumnVector;
+import org.apache.flink.table.data.vector.VectorizedColumnBatch;
+import org.apache.flink.table.data.vector.writable.WritableColumnVector;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
+import org.apache.flink.util.FlinkRuntimeException;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.parquet.column.ColumnDescriptor;
+import org.apache.parquet.column.page.PageReadStore;
+import org.apache.parquet.filter2.compat.FilterCompat;
+import org.apache.parquet.hadoop.ParquetFileReader;
+import org.apache.parquet.hadoop.metadata.BlockMetaData;
+import org.apache.parquet.hadoop.metadata.ParquetMetadata;
+import org.apache.parquet.schema.GroupType;
+import org.apache.parquet.schema.MessageType;
+import org.apache.parquet.schema.Type;
+import org.apache.parquet.schema.Types;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+
+import static org.apache.hudi.source.format.cow.ParquetSplitReaderUtil.createColumnReader;
+import static org.apache.hudi.source.format.cow.ParquetSplitReaderUtil.createWritableColumnVector;
+import static org.apache.parquet.filter2.compat.RowGroupFilter.filterRowGroups;
+import static org.apache.parquet.format.converter.ParquetMetadataConverter.range;
+import static org.apache.parquet.hadoop.ParquetFileReader.readFooter;
+import static org.apache.parquet.hadoop.ParquetInputFormat.getFilter;
+
+/**
+ * This reader is used to read a {@link VectorizedColumnBatch} from input split.
+ *
+ * <p>Note: Reference Flink release 1.11.2
+ * {@code org.apache.flink.formats.parquet.vector.ParquetColumnarRowSplitReader}
+ * because it is package scope.
+ */
+public class ParquetColumnarRowSplitReader implements Closeable {
Review comment:
wanna bring @vinothchandar into this discussion, we had this discussion before on the Spark side. Whether we should add the Engine's FileInputFormat and FileReader into Hudi's codebase. Both sides have pros and cons. @vinothchandar WDYT?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] yanghua commented on a change in pull request #2613: [HUDI-1647] Supports snapshot read for Flink
Posted by GitBox <gi...@apache.org>.
yanghua commented on a change in pull request #2613:
URL: https://github.com/apache/hudi/pull/2613#discussion_r585517728
##########
File path: hudi-flink/src/main/java/org/apache/hudi/source/format/cow/AbstractColumnReader.java
##########
@@ -0,0 +1,324 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.source.format.cow;
+
+import org.apache.flink.formats.parquet.vector.ParquetDictionary;
+import org.apache.flink.formats.parquet.vector.reader.ColumnReader;
+import org.apache.flink.table.data.vector.writable.WritableColumnVector;
+import org.apache.flink.table.data.vector.writable.WritableIntVector;
+import org.apache.parquet.Preconditions;
+import org.apache.parquet.bytes.ByteBufferInputStream;
+import org.apache.parquet.bytes.BytesInput;
+import org.apache.parquet.bytes.BytesUtils;
+import org.apache.parquet.column.ColumnDescriptor;
+import org.apache.parquet.column.Dictionary;
+import org.apache.parquet.column.Encoding;
+import org.apache.parquet.column.page.DataPage;
+import org.apache.parquet.column.page.DataPageV1;
+import org.apache.parquet.column.page.DataPageV2;
+import org.apache.parquet.column.page.DictionaryPage;
+import org.apache.parquet.column.page.PageReader;
+import org.apache.parquet.column.values.ValuesReader;
+import org.apache.parquet.io.ParquetDecodingException;
+import org.apache.parquet.schema.PrimitiveType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+
+import static org.apache.parquet.column.ValuesType.REPETITION_LEVEL;
+
+/**
+ * Abstract {@link ColumnReader}.
+ * See {@link org.apache.parquet.column.impl.ColumnReaderImpl},
+ * part of the code is referred from Apache Spark and Apache Parquet.
+ *
+ * <p>Note: Reference Flink release 1.11.2 {@link org.apache.flink.formats.parquet.vector.reader.AbstractColumnReader}
+ * because some of the package scope methods.
+ */
+public abstract class AbstractColumnReader<V extends WritableColumnVector>
+ implements ColumnReader<V> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(org.apache.flink.formats.parquet.vector.reader.AbstractColumnReader.class);
+
+ private final PageReader pageReader;
+
+ /**
+ * The dictionary, if this column has dictionary encoding.
+ */
+ protected final Dictionary dictionary;
+
+ /**
+ * Maximum definition level for this column.
+ */
+ protected final int maxDefLevel;
+
+ protected final ColumnDescriptor descriptor;
+
+ /**
+ * Total number of values read.
+ */
+ private long valuesRead;
+
+ /**
+ * value that indicates the end of the current page. That is, if valuesRead ==
+ * endOfPageValueCount, we are at the end of the page.
+ */
+ private long endOfPageValueCount;
+
+ /**
+ * If true, the current page is dictionary encoded.
+ */
+ private boolean isCurrentPageDictionaryEncoded;
+
+ /**
+ * Total values in the current page.
+ */
+ private int pageValueCount;
+
+ /*
+ * Input streams:
+ * 1.Run length encoder to encode every data, so we have run length stream to get
+ * run length information.
+ * 2.Data maybe is real data, maybe is dictionary ids which need be decode to real
+ * data from Dictionary.
+ *
+ * Run length stream ------> Data stream
+ * |
+ * ------> Dictionary ids stream
+ */
+
+ /**
+ * Run length decoder for data and dictionary.
+ */
+ protected RunLengthDecoder runLenDecoder;
+
+ /**
+ * Data input stream.
+ */
+ ByteBufferInputStream dataInputStream;
+
+ /**
+ * Dictionary decoder to wrap dictionary ids input stream.
+ */
+ private RunLengthDecoder dictionaryIdsDecoder;
+
+ public AbstractColumnReader(
+ ColumnDescriptor descriptor,
+ PageReader pageReader) throws IOException {
+ this.descriptor = descriptor;
+ this.pageReader = pageReader;
+ this.maxDefLevel = descriptor.getMaxDefinitionLevel();
+
+ DictionaryPage dictionaryPage = pageReader.readDictionaryPage();
+ if (dictionaryPage != null) {
+ try {
+ this.dictionary = dictionaryPage.getEncoding().initDictionary(descriptor, dictionaryPage);
+ this.isCurrentPageDictionaryEncoded = true;
+ } catch (IOException e) {
+ throw new IOException("could not decode the dictionary for " + descriptor, e);
+ }
+ } else {
+ this.dictionary = null;
+ this.isCurrentPageDictionaryEncoded = false;
+ }
+ /*
+ * Total number of values in this column (in this row group).
+ */
+ long totalValueCount = pageReader.getTotalValueCount();
+ if (totalValueCount == 0) {
+ throw new IOException("totalValueCount == 0");
+ }
+ }
+
+ protected void checkTypeName(PrimitiveType.PrimitiveTypeName expectedName) {
+ PrimitiveType.PrimitiveTypeName actualName = descriptor.getPrimitiveType().getPrimitiveTypeName();
+ Preconditions.checkArgument(
+ actualName == expectedName,
+ "Expected type name: %s, actual type name: %s",
+ expectedName,
+ actualName);
+ }
+
+ /**
+ * Reads `total` values from this columnReader into column.
+ */
+ @Override
+ public final void readToVector(int readNumber, V vector) throws IOException {
+ int rowId = 0;
+ WritableIntVector dictionaryIds = null;
+ if (dictionary != null) {
+ dictionaryIds = vector.reserveDictionaryIds(readNumber);
+ }
+ while (readNumber > 0) {
+ // Compute the number of values we want to read in this page.
+ int leftInPage = (int) (endOfPageValueCount - valuesRead);
+ if (leftInPage == 0) {
+ DataPage page = pageReader.readPage();
+ if (page instanceof DataPageV1) {
+ readPageV1((DataPageV1) page);
+ } else if (page instanceof DataPageV2) {
+ readPageV2((DataPageV2) page);
+ } else {
+ throw new RuntimeException("Unsupported page type: " + page.getClass());
+ }
+ leftInPage = (int) (endOfPageValueCount - valuesRead);
+ }
+ int num = Math.min(readNumber, leftInPage);
+ if (isCurrentPageDictionaryEncoded) {
+ // Read and decode dictionary ids.
+ runLenDecoder.readDictionaryIds(
+ num, dictionaryIds, vector, rowId, maxDefLevel, this.dictionaryIdsDecoder);
+
+ if (vector.hasDictionary() || (rowId == 0 && supportLazyDecode())) {
+ // Column vector supports lazy decoding of dictionary values so just set the dictionary.
+ // We can't do this if rowId != 0 AND the column doesn't have a dictionary (i.e. some
+ // non-dictionary encoded values have already been added).
+ vector.setDictionary(new ParquetDictionary(dictionary));
+ } else {
+ readBatchFromDictionaryIds(rowId, num, vector, dictionaryIds);
+ }
+ } else {
+ if (vector.hasDictionary() && rowId != 0) {
+ // This batch already has dictionary encoded values but this new page is not. The batch
+ // does not support a mix of dictionary and not so we will decode the dictionary.
+ readBatchFromDictionaryIds(0, rowId, vector, vector.getDictionaryIds());
+ }
+ vector.setDictionary(null);
+ readBatch(rowId, num, vector);
+ }
+
+ valuesRead += num;
+ rowId += num;
+ readNumber -= num;
+ }
+ }
+
+ private void readPageV1(DataPageV1 page) throws IOException {
+ this.pageValueCount = page.getValueCount();
+ ValuesReader rlReader = page.getRlEncoding().getValuesReader(descriptor, REPETITION_LEVEL);
+
+ // Initialize the decoders.
+ if (page.getDlEncoding() != Encoding.RLE && descriptor.getMaxDefinitionLevel() != 0) {
+ throw new UnsupportedOperationException("Unsupported encoding: " + page.getDlEncoding());
+ }
+ int bitWidth = BytesUtils.getWidthFromMaxInt(descriptor.getMaxDefinitionLevel());
+ this.runLenDecoder = new RunLengthDecoder(bitWidth);
+ try {
+ BytesInput bytes = page.getBytes();
+ ByteBufferInputStream in = bytes.toInputStream();
+ rlReader.initFromPage(pageValueCount, in);
+ this.runLenDecoder.initFromStream(pageValueCount, in);
+ prepareNewPage(page.getValueEncoding(), in);
+ } catch (IOException e) {
+ throw new IOException("could not read page " + page + " in col " + descriptor, e);
+ }
+ }
+
+ private void readPageV2(DataPageV2 page) throws IOException {
+ this.pageValueCount = page.getValueCount();
+
+ int bitWidth = BytesUtils.getWidthFromMaxInt(descriptor.getMaxDefinitionLevel());
+ // do not read the length from the stream. v2 pages handle dividing the page bytes.
+ this.runLenDecoder = new RunLengthDecoder(bitWidth, false);
+ this.runLenDecoder.initFromStream(
+ this.pageValueCount, page.getDefinitionLevels().toInputStream());
+ try {
+ prepareNewPage(page.getDataEncoding(), page.getData().toInputStream());
+ } catch (IOException e) {
+ throw new IOException("could not read page " + page + " in col " + descriptor, e);
+ }
+ }
+
+ private void prepareNewPage(
+ Encoding dataEncoding,
+ ByteBufferInputStream in) throws IOException {
+ this.endOfPageValueCount = valuesRead + pageValueCount;
+ if (dataEncoding.usesDictionary()) {
+ if (dictionary == null) {
+ throw new IOException("Could not read page in col "
+ + descriptor
+ + " as the dictionary was missing for encoding "
+ + dataEncoding);
+ }
+ @SuppressWarnings("deprecation")
+ Encoding plainDict = Encoding.PLAIN_DICTIONARY; // var to allow warning suppression
+ if (dataEncoding != plainDict && dataEncoding != Encoding.RLE_DICTIONARY) {
+ throw new UnsupportedOperationException("Unsupported encoding: " + dataEncoding);
+ }
+ this.dataInputStream = null;
+ this.dictionaryIdsDecoder = new RunLengthDecoder();
+ try {
+ this.dictionaryIdsDecoder.initFromStream(pageValueCount, in);
+ } catch (IOException e) {
+ throw new IOException("could not read dictionary in col " + descriptor, e);
+ }
+ this.isCurrentPageDictionaryEncoded = true;
+ } else {
+ if (dataEncoding != Encoding.PLAIN) {
+ throw new UnsupportedOperationException("Unsupported encoding: " + dataEncoding);
+ }
+ this.dictionaryIdsDecoder = null;
+ LOG.debug("init from page at offset {} for length {}", in.position(), in.available());
+ this.dataInputStream = in.remainingStream();
+ this.isCurrentPageDictionaryEncoded = false;
+ }
+
+ afterReadPage();
+ }
+
+ final ByteBuffer readDataBuffer(int length) {
+ try {
+ return dataInputStream.slice(length).order(ByteOrder.LITTLE_ENDIAN);
+ } catch (IOException e) {
+ throw new ParquetDecodingException("Failed to read " + length + " bytes", e);
+ }
+ }
+
+ /**
+ * After read a page, we may need some initialization.
+ */
+ protected void afterReadPage() {}
+
+ /**
+ * Support lazy dictionary ids decode. See more in {@link ParquetDictionary}.
+ * If return false, we will decode all the data first.
+ */
+ protected boolean supportLazyDecode() {
+ return true;
+ }
+
+ /**
+ * Read batch from {@link #runLenDecoder} and {@link #dataInputStream}.
+ */
+ protected abstract void readBatch(int rowId, int num, V column);
+
+ /**
+ * Decode dictionary ids to data.
+ * From {@link #runLenDecoder} and {@link #dictionaryIdsDecoder}.
+ */
+ protected abstract void readBatchFromDictionaryIds(
+ int rowId,
Review comment:
But compared with the `readBatch ` above this method. Why different methods follow different styles in the same class file? If we will keep the code in Hudi's codebase for a long time, we should not use different code styles.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] danny0405 commented on a change in pull request #2613: [HUDI-1647] Supports snapshot read for Flink
Posted by GitBox <gi...@apache.org>.
danny0405 commented on a change in pull request #2613:
URL: https://github.com/apache/hudi/pull/2613#discussion_r585513761
##########
File path: hudi-flink/src/main/java/org/apache/hudi/source/format/cow/AbstractColumnReader.java
##########
@@ -0,0 +1,324 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.source.format.cow;
+
+import org.apache.flink.formats.parquet.vector.ParquetDictionary;
+import org.apache.flink.formats.parquet.vector.reader.ColumnReader;
+import org.apache.flink.table.data.vector.writable.WritableColumnVector;
+import org.apache.flink.table.data.vector.writable.WritableIntVector;
+import org.apache.parquet.Preconditions;
+import org.apache.parquet.bytes.ByteBufferInputStream;
+import org.apache.parquet.bytes.BytesInput;
+import org.apache.parquet.bytes.BytesUtils;
+import org.apache.parquet.column.ColumnDescriptor;
+import org.apache.parquet.column.Dictionary;
+import org.apache.parquet.column.Encoding;
+import org.apache.parquet.column.page.DataPage;
+import org.apache.parquet.column.page.DataPageV1;
+import org.apache.parquet.column.page.DataPageV2;
+import org.apache.parquet.column.page.DictionaryPage;
+import org.apache.parquet.column.page.PageReader;
+import org.apache.parquet.column.values.ValuesReader;
+import org.apache.parquet.io.ParquetDecodingException;
+import org.apache.parquet.schema.PrimitiveType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+
+import static org.apache.parquet.column.ValuesType.REPETITION_LEVEL;
+
+/**
+ * Abstract {@link ColumnReader}.
+ * See {@link org.apache.parquet.column.impl.ColumnReaderImpl},
+ * part of the code is referred from Apache Spark and Apache Parquet.
+ *
+ * <p>Note: Reference Flink release 1.11.2 {@link org.apache.flink.formats.parquet.vector.reader.AbstractColumnReader}
+ * because some of the package scope methods.
+ */
+public abstract class AbstractColumnReader<V extends WritableColumnVector>
+ implements ColumnReader<V> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(org.apache.flink.formats.parquet.vector.reader.AbstractColumnReader.class);
+
+ private final PageReader pageReader;
+
+ /**
+ * The dictionary, if this column has dictionary encoding.
+ */
+ protected final Dictionary dictionary;
+
+ /**
+ * Maximum definition level for this column.
+ */
+ protected final int maxDefLevel;
+
+ protected final ColumnDescriptor descriptor;
+
+ /**
+ * Total number of values read.
+ */
+ private long valuesRead;
+
+ /**
+ * value that indicates the end of the current page. That is, if valuesRead ==
+ * endOfPageValueCount, we are at the end of the page.
+ */
+ private long endOfPageValueCount;
+
+ /**
+ * If true, the current page is dictionary encoded.
+ */
+ private boolean isCurrentPageDictionaryEncoded;
+
+ /**
+ * Total values in the current page.
+ */
+ private int pageValueCount;
+
+ /*
+ * Input streams:
+ * 1.Run length encoder to encode every data, so we have run length stream to get
+ * run length information.
+ * 2.Data maybe is real data, maybe is dictionary ids which need be decode to real
+ * data from Dictionary.
+ *
+ * Run length stream ------> Data stream
+ * |
+ * ------> Dictionary ids stream
+ */
+
+ /**
+ * Run length decoder for data and dictionary.
+ */
+ protected RunLengthDecoder runLenDecoder;
+
+ /**
+ * Data input stream.
+ */
+ ByteBufferInputStream dataInputStream;
+
+ /**
+ * Dictionary decoder to wrap dictionary ids input stream.
+ */
+ private RunLengthDecoder dictionaryIdsDecoder;
+
+ public AbstractColumnReader(
+ ColumnDescriptor descriptor,
+ PageReader pageReader) throws IOException {
+ this.descriptor = descriptor;
+ this.pageReader = pageReader;
+ this.maxDefLevel = descriptor.getMaxDefinitionLevel();
+
+ DictionaryPage dictionaryPage = pageReader.readDictionaryPage();
+ if (dictionaryPage != null) {
+ try {
+ this.dictionary = dictionaryPage.getEncoding().initDictionary(descriptor, dictionaryPage);
+ this.isCurrentPageDictionaryEncoded = true;
+ } catch (IOException e) {
+ throw new IOException("could not decode the dictionary for " + descriptor, e);
+ }
+ } else {
+ this.dictionary = null;
+ this.isCurrentPageDictionaryEncoded = false;
+ }
+ /*
+ * Total number of values in this column (in this row group).
+ */
+ long totalValueCount = pageReader.getTotalValueCount();
+ if (totalValueCount == 0) {
+ throw new IOException("totalValueCount == 0");
+ }
+ }
+
+ protected void checkTypeName(PrimitiveType.PrimitiveTypeName expectedName) {
+ PrimitiveType.PrimitiveTypeName actualName = descriptor.getPrimitiveType().getPrimitiveTypeName();
+ Preconditions.checkArgument(
+ actualName == expectedName,
+ "Expected type name: %s, actual type name: %s",
+ expectedName,
+ actualName);
+ }
+
+ /**
+ * Reads `total` values from this columnReader into column.
+ */
+ @Override
+ public final void readToVector(int readNumber, V vector) throws IOException {
+ int rowId = 0;
+ WritableIntVector dictionaryIds = null;
+ if (dictionary != null) {
+ dictionaryIds = vector.reserveDictionaryIds(readNumber);
+ }
+ while (readNumber > 0) {
+ // Compute the number of values we want to read in this page.
+ int leftInPage = (int) (endOfPageValueCount - valuesRead);
+ if (leftInPage == 0) {
+ DataPage page = pageReader.readPage();
+ if (page instanceof DataPageV1) {
+ readPageV1((DataPageV1) page);
+ } else if (page instanceof DataPageV2) {
+ readPageV2((DataPageV2) page);
+ } else {
+ throw new RuntimeException("Unsupported page type: " + page.getClass());
+ }
+ leftInPage = (int) (endOfPageValueCount - valuesRead);
+ }
+ int num = Math.min(readNumber, leftInPage);
+ if (isCurrentPageDictionaryEncoded) {
+ // Read and decode dictionary ids.
+ runLenDecoder.readDictionaryIds(
+ num, dictionaryIds, vector, rowId, maxDefLevel, this.dictionaryIdsDecoder);
+
+ if (vector.hasDictionary() || (rowId == 0 && supportLazyDecode())) {
+ // Column vector supports lazy decoding of dictionary values so just set the dictionary.
+ // We can't do this if rowId != 0 AND the column doesn't have a dictionary (i.e. some
+ // non-dictionary encoded values have already been added).
+ vector.setDictionary(new ParquetDictionary(dictionary));
+ } else {
+ readBatchFromDictionaryIds(rowId, num, vector, dictionaryIds);
+ }
+ } else {
+ if (vector.hasDictionary() && rowId != 0) {
+ // This batch already has dictionary encoded values but this new page is not. The batch
+ // does not support a mix of dictionary and not so we will decode the dictionary.
+ readBatchFromDictionaryIds(0, rowId, vector, vector.getDictionaryIds());
+ }
+ vector.setDictionary(null);
+ readBatch(rowId, num, vector);
+ }
+
+ valuesRead += num;
+ rowId += num;
+ readNumber -= num;
+ }
+ }
+
+ private void readPageV1(DataPageV1 page) throws IOException {
+ this.pageValueCount = page.getValueCount();
+ ValuesReader rlReader = page.getRlEncoding().getValuesReader(descriptor, REPETITION_LEVEL);
+
+ // Initialize the decoders.
+ if (page.getDlEncoding() != Encoding.RLE && descriptor.getMaxDefinitionLevel() != 0) {
+ throw new UnsupportedOperationException("Unsupported encoding: " + page.getDlEncoding());
+ }
+ int bitWidth = BytesUtils.getWidthFromMaxInt(descriptor.getMaxDefinitionLevel());
+ this.runLenDecoder = new RunLengthDecoder(bitWidth);
+ try {
+ BytesInput bytes = page.getBytes();
+ ByteBufferInputStream in = bytes.toInputStream();
+ rlReader.initFromPage(pageValueCount, in);
+ this.runLenDecoder.initFromStream(pageValueCount, in);
+ prepareNewPage(page.getValueEncoding(), in);
+ } catch (IOException e) {
+ throw new IOException("could not read page " + page + " in col " + descriptor, e);
+ }
+ }
+
+ private void readPageV2(DataPageV2 page) throws IOException {
+ this.pageValueCount = page.getValueCount();
+
+ int bitWidth = BytesUtils.getWidthFromMaxInt(descriptor.getMaxDefinitionLevel());
+ // do not read the length from the stream. v2 pages handle dividing the page bytes.
+ this.runLenDecoder = new RunLengthDecoder(bitWidth, false);
+ this.runLenDecoder.initFromStream(
+ this.pageValueCount, page.getDefinitionLevels().toInputStream());
+ try {
+ prepareNewPage(page.getDataEncoding(), page.getData().toInputStream());
+ } catch (IOException e) {
+ throw new IOException("could not read page " + page + " in col " + descriptor, e);
+ }
+ }
+
+ private void prepareNewPage(
+ Encoding dataEncoding,
+ ByteBufferInputStream in) throws IOException {
+ this.endOfPageValueCount = valuesRead + pageValueCount;
+ if (dataEncoding.usesDictionary()) {
+ if (dictionary == null) {
+ throw new IOException("Could not read page in col "
+ + descriptor
+ + " as the dictionary was missing for encoding "
+ + dataEncoding);
+ }
+ @SuppressWarnings("deprecation")
+ Encoding plainDict = Encoding.PLAIN_DICTIONARY; // var to allow warning suppression
+ if (dataEncoding != plainDict && dataEncoding != Encoding.RLE_DICTIONARY) {
+ throw new UnsupportedOperationException("Unsupported encoding: " + dataEncoding);
+ }
+ this.dataInputStream = null;
+ this.dictionaryIdsDecoder = new RunLengthDecoder();
+ try {
+ this.dictionaryIdsDecoder.initFromStream(pageValueCount, in);
+ } catch (IOException e) {
+ throw new IOException("could not read dictionary in col " + descriptor, e);
+ }
+ this.isCurrentPageDictionaryEncoded = true;
+ } else {
+ if (dataEncoding != Encoding.PLAIN) {
+ throw new UnsupportedOperationException("Unsupported encoding: " + dataEncoding);
+ }
+ this.dictionaryIdsDecoder = null;
+ LOG.debug("init from page at offset {} for length {}", in.position(), in.available());
+ this.dataInputStream = in.remainingStream();
+ this.isCurrentPageDictionaryEncoded = false;
+ }
+
+ afterReadPage();
+ }
+
+ final ByteBuffer readDataBuffer(int length) {
+ try {
+ return dataInputStream.slice(length).order(ByteOrder.LITTLE_ENDIAN);
+ } catch (IOException e) {
+ throw new ParquetDecodingException("Failed to read " + length + " bytes", e);
+ }
+ }
+
+ /**
+ * After read a page, we may need some initialization.
+ */
+ protected void afterReadPage() {}
+
+ /**
+ * Support lazy dictionary ids decode. See more in {@link ParquetDictionary}.
+ * If return false, we will decode all the data first.
+ */
+ protected boolean supportLazyDecode() {
+ return true;
+ }
+
+ /**
+ * Read batch from {@link #runLenDecoder} and {@link #dataInputStream}.
+ */
+ protected abstract void readBatch(int rowId, int num, V column);
+
+ /**
+ * Decode dictionary ids to data.
+ * From {@link #runLenDecoder} and {@link #dictionaryIdsDecoder}.
+ */
+ protected abstract void readBatchFromDictionaryIds(
+ int rowId,
Review comment:
No, i would rather keep it as is now because they are copied code.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org