You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by GitBox <gi...@apache.org> on 2020/11/25 05:01:39 UTC
[GitHub] [iceberg] zhangjun0x01 opened a new pull request #1822: Flink: add limit pushdown for IcebergTableSource
zhangjun0x01 opened a new pull request #1822:
URL: https://github.com/apache/iceberg/pull/1822
related to #1816
----------------------------------------------------------------
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
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] zhangjun0x01 commented on a change in pull request #1822: Flink: add limit pushdown for IcebergTableSource
Posted by GitBox <gi...@apache.org>.
zhangjun0x01 commented on a change in pull request #1822:
URL: https://github.com/apache/iceberg/pull/1822#discussion_r535795816
##########
File path: flink/src/main/java/org/apache/iceberg/flink/source/ScanContext.java
##########
@@ -61,6 +61,8 @@
private static final ConfigOption<Long> SPLIT_FILE_OPEN_COST =
ConfigOptions.key("split-file-open-cost").longType().defaultValue(null);
+ private static final ConfigOption<Long> LIMIT = ConfigOptions.key("limit").longType().defaultValue(-1L);
Review comment:
I removed it
----------------------------------------------------------------
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
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on pull request #1822: Flink: add limit pushdown for IcebergTableSource
Posted by GitBox <gi...@apache.org>.
openinx commented on pull request #1822:
URL: https://github.com/apache/iceberg/pull/1822#issuecomment-742225286
Merged this PR, thanks @zhangjun0x01 for contributing.
----------------------------------------------------------------
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
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] zhangjun0x01 commented on a change in pull request #1822: Flink: add limit pushdown for IcebergTableSource
Posted by GitBox <gi...@apache.org>.
zhangjun0x01 commented on a change in pull request #1822:
URL: https://github.com/apache/iceberg/pull/1822#discussion_r535795073
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSource.java
##########
@@ -0,0 +1,88 @@
+/*
+ * 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.iceberg.flink;
+
+
+import java.util.List;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import static org.junit.Assert.assertTrue;
+
+@RunWith(Parameterized.class)
+public class TestFlinkTableSource extends FlinkCatalogTestBase {
+ private static final String TABLE_NAME = "test_table";
+
+ private final FileFormat format;
+
+ @Parameterized.Parameters(name = "catalogName={0}, baseNamespace={1}, format={2}")
+ public static Iterable<Object[]> parameters() {
+ List<Object[]> parameters = Lists.newArrayList();
+ for (FileFormat format : new FileFormat[] {FileFormat.ORC, FileFormat.AVRO, FileFormat.PARQUET}) {
+ for (Object[] catalogParams : FlinkCatalogTestBase.parameters()) {
+ String catalogName = (String) catalogParams[0];
+ String[] baseNamespace = (String[]) catalogParams[1];
+ parameters.add(new Object[] {catalogName, baseNamespace, format});
+ }
+ }
+ return parameters;
+ }
+
+ public TestFlinkTableSource(String catalogName, String[] baseNamespace, FileFormat format) {
+ super(catalogName, baseNamespace);
+ this.format = format;
+ }
+
+ @Before
+ public void before() {
+ super.before();
+ sql("CREATE DATABASE %s", flinkDatabase);
+ sql("USE CATALOG %s", catalogName);
+ sql("USE %s", DATABASE);
+ sql("CREATE TABLE %s (id INT, data VARCHAR) with ('write.format.default'='%s')", TABLE_NAME, format.name());
+ }
+
+ @After
+ public void clean() {
+ sql("DROP TABLE IF EXISTS %s.%s", flinkDatabase, TABLE_NAME);
+ sql("DROP DATABASE IF EXISTS %s", flinkDatabase);
+ super.clean();
+ }
+
+ @Test
+ public void testLimitPushDown() {
+ sql("INSERT INTO %s VALUES (1,'a'),(2,'b')", TABLE_NAME);
+
+ String querySql = String.format("SELECT * FROM %s LIMIT 1", TABLE_NAME);
Review comment:
I added the test case
----------------------------------------------------------------
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
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #1822: Flink: add limit pushdown for IcebergTableSource
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1822:
URL: https://github.com/apache/iceberg/pull/1822#discussion_r533180747
##########
File path: flink/src/main/java/org/apache/iceberg/flink/IcebergTableSource.java
##########
@@ -36,25 +36,30 @@
/**
* Flink Iceberg table source.
- * TODO: Implement {@link FilterableTableSource} and {@link LimitableTableSource}.
+ * TODO: Implement {@link FilterableTableSource}
*/
-public class IcebergTableSource implements StreamTableSource<RowData>, ProjectableTableSource<RowData> {
+public class IcebergTableSource
+ implements StreamTableSource<RowData>, ProjectableTableSource<RowData>, LimitableTableSource<RowData> {
private final TableLoader loader;
private final TableSchema schema;
private final Map<String, String> properties;
private final int[] projectedFields;
+ private boolean isLimitPushDown = false;
+ private long limit = -1L;
Review comment:
if `limit` is `-1L`, then means we've disabled the limit push down , right ? If so, why do we need two fields ?
----------------------------------------------------------------
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
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #1822: Flink: add limit pushdown for IcebergTableSource
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1822:
URL: https://github.com/apache/iceberg/pull/1822#discussion_r533186123
##########
File path: flink/src/main/java/org/apache/iceberg/flink/IcebergTableSource.java
##########
@@ -102,6 +107,21 @@ public String explainSource() {
if (projectedFields != null) {
explain += ", ProjectedFields: " + Arrays.toString(projectedFields);
}
+
+ if (isLimitPushDown) {
+ explain += String.format(", LimitPushDown %s, Limit %d", isLimitPushDown, limit);
Review comment:
Here the `isLimitPushDown` is always `true`, how about use the string `String.format(", LimitPushDown: %d", limit)` ?
----------------------------------------------------------------
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
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #1822: Flink: add limit pushdown for IcebergTableSource
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1822:
URL: https://github.com/apache/iceberg/pull/1822#discussion_r533209354
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSource.java
##########
@@ -0,0 +1,88 @@
+/*
+ * 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.iceberg.flink;
+
+
+import java.util.List;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import static org.junit.Assert.assertTrue;
+
+@RunWith(Parameterized.class)
+public class TestFlinkTableSource extends FlinkCatalogTestBase {
+ private static final String TABLE_NAME = "test_table";
+
+ private final FileFormat format;
+
+ @Parameterized.Parameters(name = "catalogName={0}, baseNamespace={1}, format={2}")
+ public static Iterable<Object[]> parameters() {
+ List<Object[]> parameters = Lists.newArrayList();
+ for (FileFormat format : new FileFormat[] {FileFormat.ORC, FileFormat.AVRO, FileFormat.PARQUET}) {
+ for (Object[] catalogParams : FlinkCatalogTestBase.parameters()) {
+ String catalogName = (String) catalogParams[0];
+ String[] baseNamespace = (String[]) catalogParams[1];
+ parameters.add(new Object[] {catalogName, baseNamespace, format});
+ }
+ }
+ return parameters;
+ }
+
+ public TestFlinkTableSource(String catalogName, String[] baseNamespace, FileFormat format) {
+ super(catalogName, baseNamespace);
+ this.format = format;
+ }
+
+ @Before
+ public void before() {
+ super.before();
+ sql("CREATE DATABASE %s", flinkDatabase);
+ sql("USE CATALOG %s", catalogName);
+ sql("USE %s", DATABASE);
+ sql("CREATE TABLE %s (id INT, data VARCHAR) with ('write.format.default'='%s')", TABLE_NAME, format.name());
+ }
+
+ @After
+ public void clean() {
+ sql("DROP TABLE IF EXISTS %s.%s", flinkDatabase, TABLE_NAME);
+ sql("DROP DATABASE IF EXISTS %s", flinkDatabase);
+ super.clean();
+ }
+
+ @Test
+ public void testLimitPushDown() {
+ sql("INSERT INTO %s VALUES (1,'a'),(2,'b')", TABLE_NAME);
+
+ String querySql = String.format("SELECT * FROM %s LIMIT 1", TABLE_NAME);
Review comment:
Do we need to add those test cases:
Case.1 : `SELECT * FROM test LIMIT -1`
Case.2 : `SELECT * FROM test LIMIT 0 `
Case.3: `SELECT * FROM test LIMIT 3` , means the `limit` exceeds the total rows in table .
Case.4: `SELECT * FROM test WHERE a = 1 AND LIMIT 2` , query data with both `limit` and `filters`
etc.
----------------------------------------------------------------
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
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #1822: Flink: add limit pushdown for IcebergTableSource
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1822:
URL: https://github.com/apache/iceberg/pull/1822#discussion_r533162523
##########
File path: flink/src/main/java/org/apache/iceberg/flink/IcebergTableSource.java
##########
@@ -36,25 +36,30 @@
/**
* Flink Iceberg table source.
- * TODO: Implement {@link FilterableTableSource} and {@link LimitableTableSource}.
+ * TODO: Implement {@link FilterableTableSource}
*/
-public class IcebergTableSource implements StreamTableSource<RowData>, ProjectableTableSource<RowData> {
+public class IcebergTableSource
+ implements StreamTableSource<RowData>, ProjectableTableSource<RowData>, LimitableTableSource<RowData> {
private final TableLoader loader;
private final TableSchema schema;
private final Map<String, String> properties;
private final int[] projectedFields;
+ private boolean isLimitPushDown = false;
+ private long limit = -1L;
public IcebergTableSource(TableLoader loader, TableSchema schema, Map<String, String> properties) {
- this(loader, schema, properties, null);
+ this(loader, schema, properties, null, false, -1);
Review comment:
Why disable the limit push down for flink table source ?
----------------------------------------------------------------
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
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] zhangjun0x01 commented on a change in pull request #1822: Flink: add limit pushdown for IcebergTableSource
Posted by GitBox <gi...@apache.org>.
zhangjun0x01 commented on a change in pull request #1822:
URL: https://github.com/apache/iceberg/pull/1822#discussion_r535796008
##########
File path: flink/src/main/java/org/apache/iceberg/flink/source/FlinkInputFormat.java
##########
@@ -48,6 +48,8 @@
private final ScanContext context;
private transient RowDataIterator iterator;
+ private transient long currentReadCount = 0L;
+ private transient long limit;
Review comment:
I deleted the field `limit `
----------------------------------------------------------------
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
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #1822: Flink: add limit pushdown for IcebergTableSource
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1822:
URL: https://github.com/apache/iceberg/pull/1822#discussion_r533204460
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSource.java
##########
@@ -0,0 +1,88 @@
+/*
+ * 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.iceberg.flink;
+
+
+import java.util.List;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import static org.junit.Assert.assertTrue;
+
+@RunWith(Parameterized.class)
+public class TestFlinkTableSource extends FlinkCatalogTestBase {
+ private static final String TABLE_NAME = "test_table";
+
+ private final FileFormat format;
+
+ @Parameterized.Parameters(name = "catalogName={0}, baseNamespace={1}, format={2}")
+ public static Iterable<Object[]> parameters() {
+ List<Object[]> parameters = Lists.newArrayList();
+ for (FileFormat format : new FileFormat[] {FileFormat.ORC, FileFormat.AVRO, FileFormat.PARQUET}) {
+ for (Object[] catalogParams : FlinkCatalogTestBase.parameters()) {
+ String catalogName = (String) catalogParams[0];
+ String[] baseNamespace = (String[]) catalogParams[1];
+ parameters.add(new Object[] {catalogName, baseNamespace, format});
+ }
+ }
+ return parameters;
+ }
+
+ public TestFlinkTableSource(String catalogName, String[] baseNamespace, FileFormat format) {
+ super(catalogName, baseNamespace);
+ this.format = format;
+ }
+
+ @Before
+ public void before() {
+ super.before();
+ sql("CREATE DATABASE %s", flinkDatabase);
+ sql("USE CATALOG %s", catalogName);
+ sql("USE %s", DATABASE);
+ sql("CREATE TABLE %s (id INT, data VARCHAR) with ('write.format.default'='%s')", TABLE_NAME, format.name());
Review comment:
nit: `with` -> `WITH`
----------------------------------------------------------------
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
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #1822: Flink: add limit pushdown for IcebergTableSource
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1822:
URL: https://github.com/apache/iceberg/pull/1822#discussion_r533197314
##########
File path: flink/src/main/java/org/apache/iceberg/flink/source/FlinkInputFormat.java
##########
@@ -48,6 +48,8 @@
private final ScanContext context;
private transient RowDataIterator iterator;
+ private transient long currentReadCount = 0L;
+ private transient long limit;
Review comment:
How about just using the `context.limit()` in `reachedEnd`, rather than introducing another new transient field `limit` ?
----------------------------------------------------------------
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
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #1822: Flink: add limit pushdown for IcebergTableSource
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1822:
URL: https://github.com/apache/iceberg/pull/1822#discussion_r533193245
##########
File path: flink/src/main/java/org/apache/iceberg/flink/source/ScanContext.java
##########
@@ -61,6 +61,8 @@
private static final ConfigOption<Long> SPLIT_FILE_OPEN_COST =
ConfigOptions.key("split-file-open-cost").longType().defaultValue(null);
+ private static final ConfigOption<Long> LIMIT = ConfigOptions.key("limit").longType().defaultValue(-1L);
Review comment:
Do we need this ? we don't parse the limit from properties, right ?
----------------------------------------------------------------
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
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] zhangjun0x01 commented on a change in pull request #1822: Flink: add limit pushdown for IcebergTableSource
Posted by GitBox <gi...@apache.org>.
zhangjun0x01 commented on a change in pull request #1822:
URL: https://github.com/apache/iceberg/pull/1822#discussion_r535791036
##########
File path: flink/src/main/java/org/apache/iceberg/flink/IcebergTableSource.java
##########
@@ -36,25 +36,30 @@
/**
* Flink Iceberg table source.
- * TODO: Implement {@link FilterableTableSource} and {@link LimitableTableSource}.
+ * TODO: Implement {@link FilterableTableSource}
*/
-public class IcebergTableSource implements StreamTableSource<RowData>, ProjectableTableSource<RowData> {
+public class IcebergTableSource
+ implements StreamTableSource<RowData>, ProjectableTableSource<RowData>, LimitableTableSource<RowData> {
private final TableLoader loader;
private final TableSchema schema;
private final Map<String, String> properties;
private final int[] projectedFields;
+ private boolean isLimitPushDown = false;
+ private long limit = -1L;
Review comment:
I also think we can judge whether to pushdown by the value of limit, but the [LimitableTableSource](https://github.com/apache/flink/blob/release-1.11/flink-table/flink-table-common/src/main/java/org/apache/flink/table/sources/LimitableTableSource.java) interface provides two methods, `isLimitPushedDown` and `applyLimit`. From the method comments, I think the author wants to judge whether to pushdown by the `isLimitPushedDown` method.
In versions after flink 1.12, a new interface [SupportsLimitPushDown](https://github.com/apache/flink/blob/release-1.12/flink-table/flink-table-common/src/main/java/org/apache/flink/table/connector/source/abilities/SupportsLimitPushDown.java) is provided. This interface only provides one method. I think we can judge pushdown by the value of limit
----------------------------------------------------------------
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
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] zhangjun0x01 commented on a change in pull request #1822: Flink: add limit pushdown for IcebergTableSource
Posted by GitBox <gi...@apache.org>.
zhangjun0x01 commented on a change in pull request #1822:
URL: https://github.com/apache/iceberg/pull/1822#discussion_r535792817
##########
File path: flink/src/main/java/org/apache/iceberg/flink/IcebergTableSource.java
##########
@@ -36,25 +36,30 @@
/**
* Flink Iceberg table source.
- * TODO: Implement {@link FilterableTableSource} and {@link LimitableTableSource}.
+ * TODO: Implement {@link FilterableTableSource}
*/
-public class IcebergTableSource implements StreamTableSource<RowData>, ProjectableTableSource<RowData> {
+public class IcebergTableSource
+ implements StreamTableSource<RowData>, ProjectableTableSource<RowData>, LimitableTableSource<RowData> {
private final TableLoader loader;
private final TableSchema schema;
private final Map<String, String> properties;
private final int[] projectedFields;
+ private boolean isLimitPushDown = false;
+ private long limit = -1L;
public IcebergTableSource(TableLoader loader, TableSchema schema, Map<String, String> properties) {
- this(loader, schema, properties, null);
+ this(loader, schema, properties, null, false, -1);
Review comment:
I think it may be related to the design of the `LimitableTableSource` interface in flink 1.11. I looked up some implement classes of LimitableTableSource in flink, such as [HiveTableSource](https://github.com/apache/flink/blob/release-1.11/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveTableSource.java#L123). By default, the limit pushdown is disabled
----------------------------------------------------------------
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
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx merged pull request #1822: Flink: add limit pushdown for IcebergTableSource
Posted by GitBox <gi...@apache.org>.
openinx merged pull request #1822:
URL: https://github.com/apache/iceberg/pull/1822
----------------------------------------------------------------
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
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #1822: Flink: add limit pushdown for IcebergTableSource
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1822:
URL: https://github.com/apache/iceberg/pull/1822#discussion_r533164296
##########
File path: flink/src/main/java/org/apache/iceberg/flink/IcebergTableSource.java
##########
@@ -36,25 +36,30 @@
/**
* Flink Iceberg table source.
- * TODO: Implement {@link FilterableTableSource} and {@link LimitableTableSource}.
+ * TODO: Implement {@link FilterableTableSource}
*/
-public class IcebergTableSource implements StreamTableSource<RowData>, ProjectableTableSource<RowData> {
+public class IcebergTableSource
+ implements StreamTableSource<RowData>, ProjectableTableSource<RowData>, LimitableTableSource<RowData> {
private final TableLoader loader;
private final TableSchema schema;
private final Map<String, String> properties;
private final int[] projectedFields;
+ private boolean isLimitPushDown = false;
Review comment:
Could we add the `final` modifier ? Also no need to initialize it with a default `false` because the constructor will always assign a given value to it.
----------------------------------------------------------------
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
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #1822: Flink: add limit pushdown for IcebergTableSource
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1822:
URL: https://github.com/apache/iceberg/pull/1822#discussion_r539827673
##########
File path: flink/src/main/java/org/apache/iceberg/flink/IcebergTableSource.java
##########
@@ -36,25 +36,30 @@
/**
* Flink Iceberg table source.
- * TODO: Implement {@link FilterableTableSource} and {@link LimitableTableSource}.
+ * TODO: Implement {@link FilterableTableSource}
*/
-public class IcebergTableSource implements StreamTableSource<RowData>, ProjectableTableSource<RowData> {
+public class IcebergTableSource
+ implements StreamTableSource<RowData>, ProjectableTableSource<RowData>, LimitableTableSource<RowData> {
private final TableLoader loader;
private final TableSchema schema;
private final Map<String, String> properties;
private final int[] projectedFields;
+ private boolean isLimitPushDown = false;
+ private long limit = -1L;
public IcebergTableSource(TableLoader loader, TableSchema schema, Map<String, String> properties) {
- this(loader, schema, properties, null);
+ this(loader, schema, properties, null, false, -1);
Review comment:
Oh, the table won't have a `limit` cause by default, so we should set it disabled by default. It's OK here.
----------------------------------------------------------------
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
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org