You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "stevenzwu (via GitHub)" <gi...@apache.org> on 2024/02/14 23:34:37 UTC
[PR] Spec: add task-type field to JSON serialization of file scan task. add JSON serialization for StaticDataTask. [iceberg]
stevenzwu opened a new pull request, #9728:
URL: https://github.com/apache/iceberg/pull/9728
(no comment)
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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
Re: [PR] Core: add a new task-type field to task JSON serialization. add data task JSON serialization imp. [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #9728:
URL: https://github.com/apache/iceberg/pull/9728#discussion_r1576706517
##########
core/src/test/java/org/apache/iceberg/TestFileScanTaskParser.java:
##########
@@ -33,23 +33,64 @@ public void testNullArguments() {
.isInstanceOf(IllegalArgumentException.class)
.hasMessage("Invalid file scan task: null");
- Assertions.assertThatThrownBy(() -> FileScanTaskParser.fromJson(null, true))
+ Assertions.assertThatThrownBy(() -> FileScanTaskParser.fromJson((String) null, true))
.isInstanceOf(IllegalArgumentException.class)
.hasMessage("Invalid JSON string for file scan task: null");
+
+ Assertions.assertThatThrownBy(() -> ScanTaskParser.toJson(null))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("Invalid scan task: null");
+
+ Assertions.assertThatThrownBy(() -> ScanTaskParser.fromJson(null, true))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("Invalid JSON string for scan task: null");
}
@ParameterizedTest
@ValueSource(booleans = {true, false})
- public void testParser(boolean caseSensitive) {
+ public void testFileScanTaskParser(boolean caseSensitive) {
PartitionSpec spec = TestBase.SPEC;
- FileScanTask fileScanTask = createScanTask(spec, caseSensitive);
+ FileScanTask fileScanTask = createFileScanTask(spec, caseSensitive);
String jsonStr = FileScanTaskParser.toJson(fileScanTask);
- Assertions.assertThat(jsonStr).isEqualTo(expectedFileScanTaskJson());
+ Assertions.assertThat(jsonStr).isEqualTo(fileScanTaskJsonWithoutTaskType());
FileScanTask deserializedTask = FileScanTaskParser.fromJson(jsonStr, caseSensitive);
assertFileScanTaskEquals(fileScanTask, deserializedTask, spec, caseSensitive);
}
- private FileScanTask createScanTask(PartitionSpec spec, boolean caseSensitive) {
+ /** Test backward compatibility where task-type field is absent from the JSON string */
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testFileScanTaskParserWithoutTaskTypeField(boolean caseSensitive) {
+ PartitionSpec spec = TestBase.SPEC;
+ FileScanTask fileScanTask = createFileScanTask(spec, caseSensitive);
+ FileScanTask deserializedTask =
+ FileScanTaskParser.fromJson(fileScanTaskJsonWithoutTaskType(), caseSensitive);
+ assertFileScanTaskEquals(fileScanTask, deserializedTask, spec, caseSensitive);
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testScanTaskParser(boolean caseSensitive) {
Review Comment:
`ScanTaskParser` is the facade/dispatch class. I don't want to add all task types testing into a single gigantic test class `TestScanTaskParser`. I think it is a bit cleaner to have one test class for each task type (file, data, manifest 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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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
Re: [PR] Core: add a new task-type field to task JSON serialization. add data task JSON serialization imp. [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #9728:
URL: https://github.com/apache/iceberg/pull/9728#discussion_r1576699869
##########
core/src/main/java/org/apache/iceberg/DataTaskParser.java:
##########
@@ -0,0 +1,81 @@
+/*
+ * 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;
+
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.databind.JsonNode;
+import java.io.IOException;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.util.JsonUtil;
+
+class DataTaskParser {
+ private static final String SCHEMA = "schema";
+ private static final String PROJECTED_SCHEMA = "projection";
+ private static final String METADATA_FILE = "metadata-file";
+ private static final String ROWS = "rows";
+
+ private DataTaskParser() {}
+
+ static void toJson(StaticDataTask dataTask, JsonGenerator generator) throws IOException {
+ Preconditions.checkArgument(dataTask != null, "Invalid data task: null");
+ Preconditions.checkArgument(generator != null, "Invalid JSON generator: null");
+
+ generator.writeFieldName(SCHEMA);
+ SchemaParser.toJson(dataTask.schema(), generator);
+
+ generator.writeFieldName(PROJECTED_SCHEMA);
+ SchemaParser.toJson(dataTask.projectedSchema(), generator);
+
+ generator.writeFieldName(METADATA_FILE);
+ ContentFileParser.toJson(dataTask.metadataFile(), PartitionSpec.unpartitioned(), generator);
+
+ Preconditions.checkArgument(dataTask.tableRows() != null, "Invalid data task: null table rows");
Review Comment:
not sure if I would agree. the Preconditions on the top of the method are for input args directly. this is for field from the `dataTask` object. This is consistent with other code like `TableMetadataParser`
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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
Re: [PR] Spec, Core: add a new task-type field to task JSON serialization. add data task JSON serialization spec and imp. [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #9728:
URL: https://github.com/apache/iceberg/pull/9728#discussion_r1493034865
##########
core/src/main/java/org/apache/iceberg/FileScanTaskParser.java:
##########
@@ -21,116 +21,84 @@
import com.fasterxml.jackson.core.JsonGenerator;
import com.fasterxml.jackson.databind.JsonNode;
import java.io.IOException;
-import org.apache.iceberg.expressions.Expression;
-import org.apache.iceberg.expressions.ExpressionParser;
-import org.apache.iceberg.expressions.Expressions;
-import org.apache.iceberg.expressions.ResidualEvaluator;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
-import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.base.Strings;
import org.apache.iceberg.util.JsonUtil;
public class FileScanTaskParser {
- private static final String SCHEMA = "schema";
- private static final String SPEC = "spec";
- private static final String DATA_FILE = "data-file";
- private static final String START = "start";
- private static final String LENGTH = "length";
- private static final String DELETE_FILES = "delete-files";
- private static final String RESIDUAL = "residual-filter";
+ private static final String TASK_TYPE = "task-type";
+
+ private enum TaskType {
Review Comment:
@jackye1995 I agree that `ScanTaskParser` is more accurate for the facade/dispatcher class here. The only problem is that `FileScanTaskParser` is a public class in core module. Technically, we are breaking compatibility.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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
Re: [PR] Spec: add task-type field to JSON serialization of file scan task. add JSON serialization for StaticDataTask. [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #9728:
URL: https://github.com/apache/iceberg/pull/9728#discussion_r1490876417
##########
format/spec.md:
##########
@@ -1237,17 +1237,37 @@ Content file (data or delete) is serialized as a JSON object according to the fo
| **`equality-ids`** |`JSON list of int: Field ids used to determine row equality in equality delete files`|`[1]`|
| **`sort-order-id`** |`JSON int`|`1`|
-### File Scan Task Serialization
-
-File scan task is serialized as a JSON object according to the following table.
-
-| Metadata field |JSON representation|Example|
-|--------------------------|--- |--- |
-| **`schema`** |`JSON object`|`See above, read schemas instead`|
-| **`spec`** |`JSON object`|`See above, read partition specs instead`|
-| **`data-file`** |`JSON object`|`See above, read content file instead`|
-| **`delete-files`** |`JSON list of objects`|`See above, read content file instead`|
-| **`residual-filter`** |`JSON object: residual filter expression`|`{"type":"eq","term":"id","value":1}`|
+### Task Serialization
+
+There could be different implementations of tasks,
+e.g., `BaseFileScanTask` and `StaticDataTask` in Java.
+A enum `task-type` field is needed to distinguish different task types.
Review Comment:
internally `task-type` can be an enum, but I don't think we need to mention this here, since the JSON field is a string. Maybe just say **A `task-type` field is used to distinguish between different task types.**
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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
Re: [PR] Spec, Core: add a new task-type field to task JSON serialization. add data task JSON serialization spec and imp. [iceberg]
Posted by "rdblue (via GitHub)" <gi...@apache.org>.
rdblue commented on code in PR #9728:
URL: https://github.com/apache/iceberg/pull/9728#discussion_r1495060431
##########
format/spec.md:
##########
@@ -1237,17 +1237,36 @@ Content file (data or delete) is serialized as a JSON object according to the fo
| **`equality-ids`** |`JSON list of int: Field ids used to determine row equality in equality delete files`|`[1]`|
| **`sort-order-id`** |`JSON int`|`1`|
-### File Scan Task Serialization
-
-File scan task is serialized as a JSON object according to the following table.
-
-| Metadata field |JSON representation|Example|
-|--------------------------|--- |--- |
-| **`schema`** |`JSON object`|`See above, read schemas instead`|
-| **`spec`** |`JSON object`|`See above, read partition specs instead`|
-| **`data-file`** |`JSON object`|`See above, read content file instead`|
-| **`delete-files`** |`JSON list of objects`|`See above, read content file instead`|
-| **`residual-filter`** |`JSON object: residual filter expression`|`{"type":"eq","term":"id","value":1}`|
+### Task Serialization
Review Comment:
I would definitely prefer it if this were _not_ part of the spec. I don't think there is any reason to require a specific JSON serialization and I was surprised to see it in the spec. It's great to have documentation on exactly what the parsers produce, but we have many parsers that are not covered by the table spec and are instead in other documents like the Puffin spec, View spec, or REST spec.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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
Re: [PR] Core: add a new task-type field to task JSON serialization. add data task JSON serialization imp. [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #9728:
URL: https://github.com/apache/iceberg/pull/9728#discussion_r1576026814
##########
core/src/test/java/org/apache/iceberg/TestFileScanTaskParser.java:
##########
@@ -33,23 +33,64 @@ public void testNullArguments() {
.isInstanceOf(IllegalArgumentException.class)
.hasMessage("Invalid file scan task: null");
- Assertions.assertThatThrownBy(() -> FileScanTaskParser.fromJson(null, true))
+ Assertions.assertThatThrownBy(() -> FileScanTaskParser.fromJson((String) null, true))
.isInstanceOf(IllegalArgumentException.class)
.hasMessage("Invalid JSON string for file scan task: null");
+
+ Assertions.assertThatThrownBy(() -> ScanTaskParser.toJson(null))
Review Comment:
this should be in `TestScanTaskParser` as it's using `ScanTaskParser`
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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
Re: [PR] Core: add a new task-type field to task JSON serialization. add data task JSON serialization imp. [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on PR #9728:
URL: https://github.com/apache/iceberg/pull/9728#issuecomment-1997136910
@stevenzwu I'll take a look either tomorrow or early next week
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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
Re: [PR] Core: add a new task-type field to task JSON serialization. add data task JSON serialization imp. [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #9728:
URL: https://github.com/apache/iceberg/pull/9728#discussion_r1526599885
##########
core/src/main/java/org/apache/iceberg/FileScanTaskParser.java:
##########
@@ -40,16 +40,32 @@ public class FileScanTaskParser {
private FileScanTaskParser() {}
+ /**
+ * Serialize file scan task to JSON string
+ *
+ * @deprecated Use {@link ScanTaskParser#toJson(FileScanTask)} instead; will be removed in 1.7.0
+ */
+ @Deprecated
public static String toJson(FileScanTask fileScanTask) {
- return JsonUtil.generate(
- generator -> FileScanTaskParser.toJson(fileScanTask, generator), false);
+ Preconditions.checkArgument(fileScanTask != null, "Invalid file scan task: null");
+ return JsonUtil.generate(generator -> toJson(fileScanTask, generator), false);
+ }
+
+ /**
+ * Deserialize file scan task from JSON string
+ *
+ * @deprecated Use {@link ScanTaskParser#fromJson(String, boolean)} instead; will be removed in
+ * 1.7.0
+ */
+ @Deprecated
+ public static FileScanTask fromJson(String json, boolean caseSensitive) {
+ Preconditions.checkArgument(json != null, "Invalid JSON string for file scan task: null");
+ return JsonUtil.parse(json, node -> fromJson(node, caseSensitive));
}
- private static void toJson(FileScanTask fileScanTask, JsonGenerator generator)
- throws IOException {
+ static void toJson(FileScanTask fileScanTask, JsonGenerator generator) throws IOException {
Preconditions.checkArgument(fileScanTask != null, "Invalid file scan task: null");
Preconditions.checkArgument(generator != null, "Invalid JSON generator: null");
- generator.writeStartObject();
Review Comment:
you are right. we should have add the start/end object in the deprecated public method
```
@Deprecated
public static String toJson(FileScanTask fileScanTask) {
```
##########
core/src/main/java/org/apache/iceberg/FileScanTaskParser.java:
##########
@@ -40,16 +40,32 @@ public class FileScanTaskParser {
private FileScanTaskParser() {}
+ /**
+ * Serialize file scan task to JSON string
+ *
+ * @deprecated Use {@link ScanTaskParser#toJson(FileScanTask)} instead; will be removed in 1.7.0
+ */
+ @Deprecated
public static String toJson(FileScanTask fileScanTask) {
- return JsonUtil.generate(
- generator -> FileScanTaskParser.toJson(fileScanTask, generator), false);
+ Preconditions.checkArgument(fileScanTask != null, "Invalid file scan task: null");
+ return JsonUtil.generate(generator -> toJson(fileScanTask, generator), false);
+ }
+
+ /**
+ * Deserialize file scan task from JSON string
+ *
+ * @deprecated Use {@link ScanTaskParser#fromJson(String, boolean)} instead; will be removed in
+ * 1.7.0
+ */
+ @Deprecated
+ public static FileScanTask fromJson(String json, boolean caseSensitive) {
+ Preconditions.checkArgument(json != null, "Invalid JSON string for file scan task: null");
+ return JsonUtil.parse(json, node -> fromJson(node, caseSensitive));
}
- private static void toJson(FileScanTask fileScanTask, JsonGenerator generator)
- throws IOException {
+ static void toJson(FileScanTask fileScanTask, JsonGenerator generator) throws IOException {
Preconditions.checkArgument(fileScanTask != null, "Invalid file scan task: null");
Preconditions.checkArgument(generator != null, "Invalid JSON generator: null");
- generator.writeStartObject();
Review Comment:
you are right. should have added the start/end object in the deprecated public method
```
@Deprecated
public static String toJson(FileScanTask fileScanTask) {
```
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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
Re: [PR] Spec: add task-type field to JSON serialization of file scan task. add JSON serialization for StaticDataTask. [iceberg]
Posted by "rdblue (via GitHub)" <gi...@apache.org>.
rdblue commented on code in PR #9728:
URL: https://github.com/apache/iceberg/pull/9728#discussion_r1490201923
##########
format/spec.md:
##########
@@ -1239,15 +1239,34 @@ Content file (data or delete) is serialized as a JSON object according to the fo
### File Scan Task Serialization
-File scan task is serialized as a JSON object according to the following table.
-
-| Metadata field |JSON representation|Example|
-|--------------------------|--- |--- |
-| **`schema`** |`JSON object`|`See above, read schemas instead`|
-| **`spec`** |`JSON object`|`See above, read partition specs instead`|
-| **`data-file`** |`JSON object`|`See above, read content file instead`|
-| **`delete-files`** |`JSON list of objects`|`See above, read content file instead`|
-| **`residual-filter`** |`JSON object: residual filter expression`|`{"type":"eq","term":"id","value":1}`|
+There could be different implementations of file scan task,
+e.g., `BaseFileScanTask` and `StaticDataTask` in Java.
+A enum `task-type` field is needed to distinguish different task types.
+
+| Metadata field | JSON representation | Example |
+|-----------------|---------------------|---------------------------------------------------------------------------------------|
+| **`task-type`** | `JSON string` | `base-file-task`, `static-data-task`. Absence of this field should be interpreted as `base-file-task` |
Review Comment:
This should not have "base" in it. That's a Java-specific prefix we use for classes that are the base implementation of an interface.
I'm not sure about having a task type here if this is a file scan task. A file scan task should be distinct from a data task. In the Java API, a data task is a file scan task, but that is just for API compatibility. They are conceptually separate so we should keep them separate here. Maybe that just means this section should be "Task Serialization" and we have a different set of fields for "file-scan-task" or "data-task"?
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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
Re: [PR] Spec, Core: add a new task-type field to task JSON serialization. add data task JSON serialization spec and imp. [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #9728:
URL: https://github.com/apache/iceberg/pull/9728#discussion_r1493037575
##########
format/spec.md:
##########
@@ -1237,17 +1237,36 @@ Content file (data or delete) is serialized as a JSON object according to the fo
| **`equality-ids`** |`JSON list of int: Field ids used to determine row equality in equality delete files`|`[1]`|
| **`sort-order-id`** |`JSON int`|`1`|
-### File Scan Task Serialization
-
-File scan task is serialized as a JSON object according to the following table.
-
-| Metadata field |JSON representation|Example|
-|--------------------------|--- |--- |
-| **`schema`** |`JSON object`|`See above, read schemas instead`|
-| **`spec`** |`JSON object`|`See above, read partition specs instead`|
-| **`data-file`** |`JSON object`|`See above, read content file instead`|
-| **`delete-files`** |`JSON list of objects`|`See above, read content file instead`|
-| **`residual-filter`** |`JSON object: residual filter expression`|`{"type":"eq","term":"id","value":1}`|
+### Task Serialization
Review Comment:
@emkornfield that is a fair question. @rdblue also had a question if this should be added to REST OpenAPI. My reservation is that this is also used by Flink for state serialization. Hence it is not just a REST thing.
what do others think? @rdblue @jackye1995 @aokolnychyi
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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
Re: [PR] Core: add a new task-type field to task JSON serialization. add data task JSON serialization imp. [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #9728:
URL: https://github.com/apache/iceberg/pull/9728#discussion_r1576036541
##########
core/src/main/java/org/apache/iceberg/ScanTaskParser.java:
##########
@@ -0,0 +1,105 @@
+/*
+ * 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;
+
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.databind.JsonNode;
+import java.io.IOException;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.base.Strings;
+import org.apache.iceberg.util.JsonUtil;
+
+public class ScanTaskParser {
+ private static final String TASK_TYPE = "task-type";
+
+ private enum TaskType {
+ FILE_SCAN_TASK("file-scan-task"),
+ DATA_TASK("data-task");
+
+ private final String value;
+
+ TaskType(String value) {
+ this.value = value;
+ }
+
+ public static TaskType fromTypeName(String value) {
+ Preconditions.checkArgument(
+ !Strings.isNullOrEmpty(value), "Invalid task type name: null or empty");
+ if (FILE_SCAN_TASK.typeName().equalsIgnoreCase(value)) {
+ return FILE_SCAN_TASK;
+ } else if (DATA_TASK.typeName().equalsIgnoreCase(value)) {
+ return DATA_TASK;
+ } else {
+ throw new IllegalArgumentException("Unknown task type: " + value);
+ }
+ }
+
+ public String typeName() {
+ return value;
+ }
+ }
+
+ private ScanTaskParser() {}
+
+ public static String toJson(FileScanTask fileScanTask) {
+ Preconditions.checkArgument(fileScanTask != null, "Invalid scan task: null");
+ return JsonUtil.generate(generator -> toJson(fileScanTask, generator), false);
+ }
+
+ public static FileScanTask fromJson(String json, boolean caseSensitive) {
+ Preconditions.checkArgument(json != null, "Invalid JSON string for scan task: null");
+ return JsonUtil.parse(json, node -> fromJson(node, caseSensitive));
+ }
+
+ private static void toJson(FileScanTask fileScanTask, JsonGenerator generator)
+ throws IOException {
+ generator.writeStartObject();
+
+ if (fileScanTask instanceof StaticDataTask) {
+ generator.writeStringField(TASK_TYPE, TaskType.DATA_TASK.typeName());
+ DataTaskParser.toJson((StaticDataTask) fileScanTask, generator);
+ } else if (fileScanTask instanceof BaseFileScanTask
+ || fileScanTask instanceof BaseFileScanTask.SplitScanTask) {
+ generator.writeStringField(TASK_TYPE, TaskType.FILE_SCAN_TASK.typeName());
+ FileScanTaskParser.toJson(fileScanTask, generator);
+ } else {
+ throw new UnsupportedOperationException(
+ "Unsupported task type: " + fileScanTask.getClass().getCanonicalName());
Review Comment:
there should probably be a test that checks for this particular error msg
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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
Re: [PR] Core: add a new task-type field to task JSON serialization. add data task JSON serialization imp. [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #9728:
URL: https://github.com/apache/iceberg/pull/9728#discussion_r1583002079
##########
core/src/test/java/org/apache/iceberg/TestDataTaskParser.java:
##########
@@ -0,0 +1,249 @@
+/*
+ * 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;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.io.StringWriter;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.util.JsonUtil;
+import org.junit.jupiter.api.Test;
+
+public class TestDataTaskParser {
+ @Test
+ public void nullCheck() throws Exception {
+ StringWriter writer = new StringWriter();
+ JsonGenerator generator = JsonUtil.factory().createGenerator(writer);
+
+ assertThatThrownBy(() -> DataTaskParser.toJson(null, generator))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("Invalid data task: null");
+
+ assertThatThrownBy(() -> DataTaskParser.toJson((StaticDataTask) createDataTask(), null))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("Invalid JSON generator: null");
+
+ assertThatThrownBy(() -> DataTaskParser.fromJson(null))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("Invalid JSON node for data task: null");
+ }
+
+ @Test
+ public void invalidJsonNode() throws Exception {
+ String jsonStr = "{\"str\":\"1\", \"arr\":[]}";
+ ObjectMapper mapper = new ObjectMapper();
+ JsonNode rootNode = mapper.reader().readTree(jsonStr);
+
+ assertThatThrownBy(() -> DataTaskParser.fromJson(rootNode.get("str")))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Invalid JSON node for data task: non-object ");
+
+ assertThatThrownBy(() -> DataTaskParser.fromJson(rootNode.get("arr")))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Invalid JSON node for data task: non-object ");
+ }
+
+ @Test
+ public void missingFields() throws Exception {
+ ObjectMapper mapper = new ObjectMapper();
+
+ String missingSchemaStr = "{}";
+ JsonNode missingSchemaNode = mapper.reader().readTree(missingSchemaStr);
+ assertThatThrownBy(() -> DataTaskParser.fromJson(missingSchemaNode))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Cannot parse missing field: schema");
+
+ String missingProjectionStr =
+ "{"
+ + "\"schema\":{\"type\":\"struct\",\"schema-id\":0,"
+ + "\"fields\":[{\"id\":1,\"name\":\"committed_at\",\"required\":true,\"type\":\"timestamptz\"}]}"
+ + "}";
+ JsonNode missingProjectionNode = mapper.reader().readTree(missingProjectionStr);
+ assertThatThrownBy(() -> DataTaskParser.fromJson(missingProjectionNode))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Cannot parse missing field: projection");
+
+ String missingMetadataFileStr =
+ "{"
+ + "\"schema\":{\"type\":\"struct\",\"schema-id\":0,"
+ + "\"fields\":[{\"id\":1,\"name\":\"committed_at\",\"required\":true,\"type\":\"timestamptz\"}]},"
+ + "\"projection\":{\"type\":\"struct\",\"schema-id\":0,"
+ + "\"fields\":[{\"id\":1,\"name\":\"committed_at\",\"required\":true,\"type\":\"timestamptz\"}]}"
+ + "}";
+ JsonNode missingMetadataFileNode = mapper.reader().readTree(missingMetadataFileStr);
+ assertThatThrownBy(() -> DataTaskParser.fromJson(missingMetadataFileNode))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Cannot parse missing field: metadata-file");
+
+ String missingTableRowsStr =
+ "{\"task-type\":\"data-task\","
+ + "\"schema\":{\"type\":\"struct\",\"schema-id\":0,"
+ + "\"fields\":[{\"id\":1,\"name\":\"committed_at\",\"required\":true,\"type\":\"timestamptz\"},"
+ + "{\"id\":2,\"name\":\"snapshot_id\",\"required\":true,\"type\":\"long\"},"
+ + "{\"id\":3,\"name\":\"parent_id\",\"required\":false,\"type\":\"long\"},"
+ + "{\"id\":4,\"name\":\"operation\",\"required\":false,\"type\":\"string\"},"
+ + "{\"id\":5,\"name\":\"manifest_list\",\"required\":false,\"type\":\"string\"},"
+ + "{\"id\":6,\"name\":\"summary\",\"required\":false,\"type\":{\"type\":\"map\","
+ + "\"key-id\":7,\"key\":\"string\",\"value-id\":8,"
+ + "\"value\":\"string\",\"value-required\":true}}]},"
+ + "\"projection\":{\"type\":\"struct\",\"schema-id\":0,"
+ + "\"fields\":[{\"id\":1,\"name\":\"committed_at\",\"required\":true,\"type\":\"timestamptz\"},"
+ + "{\"id\":2,\"name\":\"snapshot_id\",\"required\":true,\"type\":\"long\"},"
+ + "{\"id\":3,\"name\":\"parent_id\",\"required\":false,\"type\":\"long\"},"
+ + "{\"id\":4,\"name\":\"operation\",\"required\":false,\"type\":\"string\"},"
+ + "{\"id\":5,\"name\":\"manifest_list\",\"required\":false,\"type\":\"string\"},"
+ + "{\"id\":6,\"name\":\"summary\",\"required\":false,\"type\":{\"type\":\"map\","
+ + "\"key-id\":7,\"key\":\"string\",\"value-id\":8,"
+ + "\"value\":\"string\",\"value-required\":true}}]},"
+ + "\"metadata-file\":{\"spec-id\":0,\"content\":\"DATA\","
+ + "\"file-path\":\"/tmp/metadata2.json\","
+ + "\"file-format\":\"METADATA\",\"partition\":{},"
+ + "\"file-size-in-bytes\":0,\"record-count\":2,\"sort-order-id\":0}"
+ + "}";
+ JsonNode missingTableRowsNode = mapper.reader().readTree(missingTableRowsStr);
+ assertThatThrownBy(() -> DataTaskParser.fromJson(missingTableRowsNode))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Cannot parse missing field: rows");
+ }
+
+ @Test
+ public void roundTripSerde() {
+ StaticDataTask dataTask = (StaticDataTask) createDataTask();
+ String jsonStr = ScanTaskParser.toJson(dataTask);
+ assertThat(jsonStr).isEqualTo(snapshotsDataTaskJson());
+ StaticDataTask deserializedTask = (StaticDataTask) ScanTaskParser.fromJson(jsonStr, true);
+ assertDataTaskEquals(dataTask, deserializedTask);
+ }
+
+ private DataTask createDataTask() {
+ Map<String, String> summary1 =
+ ImmutableMap.of(
+ "added-data-files", "1",
+ "added-records", "1",
+ "added-files-size", "10",
+ "changed-partition-count", "1",
+ "total-records", "1",
+ "total-files-size", "10",
+ "total-data-files", "1",
+ "total-delete-files", "0",
+ "total-position-deletes", "0",
+ "total-equality-deletes", "0");
+
+ Map<String, String> summary2 =
+ ImmutableMap.of(
+ "added-data-files", "1",
+ "added-records", "1",
+ "added-files-size", "10",
+ "changed-partition-count", "1",
+ "total-records", "2",
+ "total-files-size", "20",
+ "total-data-files", "2",
+ "total-delete-files", "0",
+ "total-position-deletes", "0",
+ "total-equality-deletes", "0");
+
+ List<Snapshot> snapshots =
+ Arrays.asList(
+ new BaseSnapshot(
+ 1L, 1L, null, 1234567890000L, "append", summary1, 1, "file:/tmp/manifest1.avro"),
+ new BaseSnapshot(
+ 2L, 2L, 1L, 9876543210000L, "append", summary2, 1, "file:/tmp/manifest2.avro"));
+
+ return StaticDataTask.of(
+ Files.localInput("file:/tmp/metadata2.json"),
+ SnapshotsTable.SNAPSHOT_SCHEMA,
+ SnapshotsTable.SNAPSHOT_SCHEMA,
+ snapshots,
+ SnapshotsTable::snapshotToRow);
+ }
+
+ private String snapshotsDataTaskJson() {
+ return "{\"task-type\":\"data-task\","
+ + "\"schema\":{\"type\":\"struct\",\"schema-id\":0,"
+ + "\"fields\":[{\"id\":1,\"name\":\"committed_at\",\"required\":true,\"type\":\"timestamptz\"},"
+ + "{\"id\":2,\"name\":\"snapshot_id\",\"required\":true,\"type\":\"long\"},"
+ + "{\"id\":3,\"name\":\"parent_id\",\"required\":false,\"type\":\"long\"},"
+ + "{\"id\":4,\"name\":\"operation\",\"required\":false,\"type\":\"string\"},"
+ + "{\"id\":5,\"name\":\"manifest_list\",\"required\":false,\"type\":\"string\"},"
+ + "{\"id\":6,\"name\":\"summary\",\"required\":false,\"type\":{\"type\":\"map\","
+ + "\"key-id\":7,\"key\":\"string\",\"value-id\":8,"
+ + "\"value\":\"string\",\"value-required\":true}}]},"
+ + "\"projection\":{\"type\":\"struct\",\"schema-id\":0,"
+ + "\"fields\":[{\"id\":1,\"name\":\"committed_at\",\"required\":true,\"type\":\"timestamptz\"},"
+ + "{\"id\":2,\"name\":\"snapshot_id\",\"required\":true,\"type\":\"long\"},"
+ + "{\"id\":3,\"name\":\"parent_id\",\"required\":false,\"type\":\"long\"},"
+ + "{\"id\":4,\"name\":\"operation\",\"required\":false,\"type\":\"string\"},"
+ + "{\"id\":5,\"name\":\"manifest_list\",\"required\":false,\"type\":\"string\"},"
+ + "{\"id\":6,\"name\":\"summary\",\"required\":false,\"type\":{\"type\":\"map\","
+ + "\"key-id\":7,\"key\":\"string\",\"value-id\":8,"
+ + "\"value\":\"string\",\"value-required\":true}}]},"
+ + "\"metadata-file\":{\"spec-id\":0,\"content\":\"DATA\","
+ + "\"file-path\":\"/tmp/metadata2.json\","
+ + "\"file-format\":\"METADATA\",\"partition\":{},"
+ + "\"file-size-in-bytes\":0,\"record-count\":2,\"sort-order-id\":0},"
+ + "\"rows\":[{\"1\":\"2009-02-13T23:31:30+00:00\",\"2\":1,\"4\":\"append\","
+ + "\"5\":\"file:/tmp/manifest1.avro\","
+ + "\"6\":{\"keys\":[\"added-data-files\",\"added-records\",\"added-files-size\",\"changed-partition-count\","
+ + "\"total-records\",\"total-files-size\",\"total-data-files\",\"total-delete-files\","
+ + "\"total-position-deletes\",\"total-equality-deletes\"],"
+ + "\"values\":[\"1\",\"1\",\"10\",\"1\",\"1\",\"10\",\"1\",\"0\",\"0\",\"0\"]}},"
+ + "{\"1\":\"2282-12-22T20:13:30+00:00\",\"2\":2,\"3\":1,\"4\":\"append\","
+ + "\"5\":\"file:/tmp/manifest2.avro\","
+ + "\"6\":{\"keys\":[\"added-data-files\",\"added-records\",\"added-files-size\",\"changed-partition-count\","
+ + "\"total-records\",\"total-files-size\",\"total-data-files\",\"total-delete-files\","
+ + "\"total-position-deletes\",\"total-equality-deletes\"],"
+ + "\"values\":[\"1\",\"1\",\"10\",\"1\",\"2\",\"20\",\"2\",\"0\",\"0\",\"0\"]}}]}";
+ }
+
+ private void assertDataTaskEquals(StaticDataTask expected, StaticDataTask actual) {
+ assertThat(expected.schema().asStruct())
Review Comment:
actual/expected is the wrong way. Should be `assertThat(actual...)...isEqualTo(expected...)`. Same for the other assertions
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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
Re: [PR] Core: add a new task-type field to task JSON serialization. add data task JSON serialization imp. [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #9728:
URL: https://github.com/apache/iceberg/pull/9728#discussion_r1576014466
##########
core/src/main/java/org/apache/iceberg/DataTaskParser.java:
##########
@@ -0,0 +1,81 @@
+/*
+ * 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;
+
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.databind.JsonNode;
+import java.io.IOException;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.util.JsonUtil;
+
+class DataTaskParser {
+ private static final String SCHEMA = "schema";
+ private static final String PROJECTED_SCHEMA = "projection";
+ private static final String METADATA_FILE = "metadata-file";
+ private static final String ROWS = "rows";
+
+ private DataTaskParser() {}
+
+ static void toJson(StaticDataTask dataTask, JsonGenerator generator) throws IOException {
+ Preconditions.checkArgument(dataTask != null, "Invalid data task: null");
+ Preconditions.checkArgument(generator != null, "Invalid JSON generator: null");
+
+ generator.writeFieldName(SCHEMA);
+ SchemaParser.toJson(dataTask.schema(), generator);
+
+ generator.writeFieldName(PROJECTED_SCHEMA);
+ SchemaParser.toJson(dataTask.projectedSchema(), generator);
+
+ generator.writeFieldName(METADATA_FILE);
+ ContentFileParser.toJson(dataTask.metadataFile(), PartitionSpec.unpartitioned(), generator);
+
+ Preconditions.checkArgument(dataTask.tableRows() != null, "Invalid data task: null table rows");
Review Comment:
this can probably be moved to the top of the method, right after the other preconditions
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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
Re: [PR] Core: add a new task-type field to task JSON serialization. add data task JSON serialization imp. [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #9728:
URL: https://github.com/apache/iceberg/pull/9728#discussion_r1576017301
##########
core/src/main/java/org/apache/iceberg/FileScanTaskParser.java:
##########
@@ -40,16 +40,39 @@ public class FileScanTaskParser {
private FileScanTaskParser() {}
+ /**
+ * Serialize file scan task to JSON string
+ *
+ * @deprecated Use {@link ScanTaskParser#toJson(FileScanTask)} instead; will be removed in 1.7.0
Review Comment:
```suggestion
* @deprecated will be removed in 1.7.0; use {@link ScanTaskParser#toJson(FileScanTask)} instead
```
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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
Re: [PR] Core: add a new task-type field to task JSON serialization. add data task JSON serialization imp. [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #9728:
URL: https://github.com/apache/iceberg/pull/9728#discussion_r1576706517
##########
core/src/test/java/org/apache/iceberg/TestFileScanTaskParser.java:
##########
@@ -33,23 +33,64 @@ public void testNullArguments() {
.isInstanceOf(IllegalArgumentException.class)
.hasMessage("Invalid file scan task: null");
- Assertions.assertThatThrownBy(() -> FileScanTaskParser.fromJson(null, true))
+ Assertions.assertThatThrownBy(() -> FileScanTaskParser.fromJson((String) null, true))
.isInstanceOf(IllegalArgumentException.class)
.hasMessage("Invalid JSON string for file scan task: null");
+
+ Assertions.assertThatThrownBy(() -> ScanTaskParser.toJson(null))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("Invalid scan task: null");
+
+ Assertions.assertThatThrownBy(() -> ScanTaskParser.fromJson(null, true))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("Invalid JSON string for scan task: null");
}
@ParameterizedTest
@ValueSource(booleans = {true, false})
- public void testParser(boolean caseSensitive) {
+ public void testFileScanTaskParser(boolean caseSensitive) {
PartitionSpec spec = TestBase.SPEC;
- FileScanTask fileScanTask = createScanTask(spec, caseSensitive);
+ FileScanTask fileScanTask = createFileScanTask(spec, caseSensitive);
String jsonStr = FileScanTaskParser.toJson(fileScanTask);
- Assertions.assertThat(jsonStr).isEqualTo(expectedFileScanTaskJson());
+ Assertions.assertThat(jsonStr).isEqualTo(fileScanTaskJsonWithoutTaskType());
FileScanTask deserializedTask = FileScanTaskParser.fromJson(jsonStr, caseSensitive);
assertFileScanTaskEquals(fileScanTask, deserializedTask, spec, caseSensitive);
}
- private FileScanTask createScanTask(PartitionSpec spec, boolean caseSensitive) {
+ /** Test backward compatibility where task-type field is absent from the JSON string */
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testFileScanTaskParserWithoutTaskTypeField(boolean caseSensitive) {
+ PartitionSpec spec = TestBase.SPEC;
+ FileScanTask fileScanTask = createFileScanTask(spec, caseSensitive);
+ FileScanTask deserializedTask =
+ FileScanTaskParser.fromJson(fileScanTaskJsonWithoutTaskType(), caseSensitive);
+ assertFileScanTaskEquals(fileScanTask, deserializedTask, spec, caseSensitive);
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testScanTaskParser(boolean caseSensitive) {
Review Comment:
`ScanTaskParser` is the facade/dispatch class. I don't want to add all task types testing into a single `TestScanTaskParser`. I think it is a bit cleaner to have one test class for each task type.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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
Re: [PR] Spec, Core: add a new task-type field to task JSON serialization. add data task JSON serialization spec and imp. [iceberg]
Posted by "rdblue (via GitHub)" <gi...@apache.org>.
rdblue commented on code in PR #9728:
URL: https://github.com/apache/iceberg/pull/9728#discussion_r1495060431
##########
format/spec.md:
##########
@@ -1237,17 +1237,36 @@ Content file (data or delete) is serialized as a JSON object according to the fo
| **`equality-ids`** |`JSON list of int: Field ids used to determine row equality in equality delete files`|`[1]`|
| **`sort-order-id`** |`JSON int`|`1`|
-### File Scan Task Serialization
-
-File scan task is serialized as a JSON object according to the following table.
-
-| Metadata field |JSON representation|Example|
-|--------------------------|--- |--- |
-| **`schema`** |`JSON object`|`See above, read schemas instead`|
-| **`spec`** |`JSON object`|`See above, read partition specs instead`|
-| **`data-file`** |`JSON object`|`See above, read content file instead`|
-| **`delete-files`** |`JSON list of objects`|`See above, read content file instead`|
-| **`residual-filter`** |`JSON object: residual filter expression`|`{"type":"eq","term":"id","value":1}`|
+### Task Serialization
Review Comment:
I would definitely prefer it if this were _not_ part of the spec. I don't think there is any reason to require a specific JSON serialization and I was surprised to see it in the spec. It's great to have documentation on exactly what the parsers produce, but we have many parsers that are not covered by the table spec and are instead in other documents like the Puffin spec, View spec, or REST spec.
To me, state serialization is a concern internal to Flink. It's harder to adhere to a spec for that, plus make guarantees about forward and backward compatibility. And without context for how this is used and why it is here, we can't make decisions about how to evolve this. For example, if you wanted to remove a field that Flink doesn't use, how do you know whether that is safe in the table spec? What does it mean for this to evolve "safely"?
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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
Re: [PR] Core: add a new task-type field to task JSON serialization. add data task JSON serialization imp. [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #9728:
URL: https://github.com/apache/iceberg/pull/9728#discussion_r1524824279
##########
core/src/main/java/org/apache/iceberg/ScanTaskParser.java:
##########
@@ -0,0 +1,104 @@
+/*
+ * 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;
+
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.databind.JsonNode;
+import java.io.IOException;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.base.Strings;
+import org.apache.iceberg.util.JsonUtil;
+
+public class ScanTaskParser {
+ private static final String TASK_TYPE = "task-type";
+
+ private enum TaskType {
+ FILE_SCAN_TASK("file-scan-task"),
+ DATA_TASK("data-task");
+
+ private final String value;
+
+ TaskType(String value) {
+ this.value = value;
+ }
+
+ public static TaskType fromValue(String value) {
Review Comment:
the codebase typically names this `fromString()`
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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
Re: [PR] Spec, Core: add a new task-type field to task JSON serialization. add data task JSON serialization spec and imp. [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #9728:
URL: https://github.com/apache/iceberg/pull/9728#discussion_r1493034865
##########
core/src/main/java/org/apache/iceberg/FileScanTaskParser.java:
##########
@@ -21,116 +21,84 @@
import com.fasterxml.jackson.core.JsonGenerator;
import com.fasterxml.jackson.databind.JsonNode;
import java.io.IOException;
-import org.apache.iceberg.expressions.Expression;
-import org.apache.iceberg.expressions.ExpressionParser;
-import org.apache.iceberg.expressions.Expressions;
-import org.apache.iceberg.expressions.ResidualEvaluator;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
-import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.base.Strings;
import org.apache.iceberg.util.JsonUtil;
public class FileScanTaskParser {
- private static final String SCHEMA = "schema";
- private static final String SPEC = "spec";
- private static final String DATA_FILE = "data-file";
- private static final String START = "start";
- private static final String LENGTH = "length";
- private static final String DELETE_FILES = "delete-files";
- private static final String RESIDUAL = "residual-filter";
+ private static final String TASK_TYPE = "task-type";
+
+ private enum TaskType {
Review Comment:
@jackye1995 I agree that `ScanTaskParser` is more accurate for the facade/dispatcher class here. The only problem is that `FileScanTaskParser` is a public class in core module. Technically, we are breaking compatibility.
maybe we can keep `FileScanTaskParser` and mark it as deprecated. it can just extend from the new `ScanTaskParser` to avoid code duplication. But it means that we need to keep the file task parser name as `BaseFileScanTaskParser`, which is package private anyway.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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
Re: [PR] Core: add a new task-type field to task JSON serialization. add data task JSON serialization imp. [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #9728:
URL: https://github.com/apache/iceberg/pull/9728#discussion_r1526490459
##########
core/src/main/java/org/apache/iceberg/ScanTaskParser.java:
##########
@@ -0,0 +1,104 @@
+/*
+ * 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;
+
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.databind.JsonNode;
+import java.io.IOException;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.base.Strings;
+import org.apache.iceberg.util.JsonUtil;
+
+public class ScanTaskParser {
+ private static final String TASK_TYPE = "task-type";
+
+ private enum TaskType {
+ FILE_SCAN_TASK("file-scan-task"),
+ DATA_TASK("data-task");
+
+ private final String value;
+
+ TaskType(String value) {
+ this.value = value;
+ }
+
+ public static TaskType fromValue(String value) {
+ Preconditions.checkArgument(value != null, "Task type is null");
+ if (FILE_SCAN_TASK.value().equalsIgnoreCase(value)) {
+ return FILE_SCAN_TASK;
+ } else if (DATA_TASK.value().equalsIgnoreCase(value)) {
+ return DATA_TASK;
+ } else {
+ throw new IllegalArgumentException("Unknown task type: " + value);
+ }
+ }
+
+ public String value() {
+ return value;
+ }
+ }
+
+ private ScanTaskParser() {}
+
+ public static String toJson(FileScanTask fileScanTask) {
+ Preconditions.checkArgument(fileScanTask != null, "Invalid file scan task: null");
+ return JsonUtil.generate(generator -> toJson(fileScanTask, generator), false);
+ }
+
+ public static FileScanTask fromJson(String json, boolean caseSensitive) {
+ Preconditions.checkArgument(json != null, "Invalid JSON string for file scan task: null");
+ return JsonUtil.parse(json, node -> fromJson(node, caseSensitive));
+ }
+
+ private static void toJson(FileScanTask fileScanTask, JsonGenerator generator)
+ throws IOException {
+ generator.writeStartObject();
+
+ if (fileScanTask instanceof StaticDataTask) {
+ generator.writeStringField(TASK_TYPE, TaskType.DATA_TASK.value());
+ DataTaskParser.toJson((StaticDataTask) fileScanTask, generator);
+ } else if (fileScanTask instanceof BaseFileScanTask
+ || fileScanTask instanceof BaseFileScanTask.SplitScanTask) {
+ generator.writeStringField(TASK_TYPE, TaskType.FILE_SCAN_TASK.value());
+ FileScanTaskParser.toJson(fileScanTask, generator);
+ } else {
+ throw new UnsupportedOperationException(
Review Comment:
see my reply in the comment above
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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
Re: [PR] Core: add a new task-type field to task JSON serialization. add data task JSON serialization imp. [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #9728:
URL: https://github.com/apache/iceberg/pull/9728#discussion_r1526625686
##########
core/src/test/java/org/apache/iceberg/TestFileScanTaskParser.java:
##########
@@ -27,29 +27,41 @@
import org.junit.jupiter.params.provider.ValueSource;
public class TestFileScanTaskParser {
+
@Test
public void testNullArguments() {
- Assertions.assertThatThrownBy(() -> FileScanTaskParser.toJson(null))
+ Assertions.assertThatThrownBy(() -> ScanTaskParser.toJson(null))
Review Comment:
I see your point here. Will add a `TestScanTaskParser` to test the facade part, like task type field and unsupported task type.
You pointed out another issue in this test class. we should test both the old/deprecated public API (via `FileScanTaskParser.toJson/fromJson`) and new `ScanTaskParser` public API.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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
Re: [PR] Spec: add task-type field to JSON serialization of file scan task. add JSON serialization for StaticDataTask. [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #9728:
URL: https://github.com/apache/iceberg/pull/9728#discussion_r1490871701
##########
format/spec.md:
##########
@@ -1237,17 +1237,37 @@ Content file (data or delete) is serialized as a JSON object according to the fo
| **`equality-ids`** |`JSON list of int: Field ids used to determine row equality in equality delete files`|`[1]`|
| **`sort-order-id`** |`JSON int`|`1`|
-### File Scan Task Serialization
-
-File scan task is serialized as a JSON object according to the following table.
-
-| Metadata field |JSON representation|Example|
-|--------------------------|--- |--- |
-| **`schema`** |`JSON object`|`See above, read schemas instead`|
-| **`spec`** |`JSON object`|`See above, read partition specs instead`|
-| **`data-file`** |`JSON object`|`See above, read content file instead`|
-| **`delete-files`** |`JSON list of objects`|`See above, read content file instead`|
-| **`residual-filter`** |`JSON object: residual filter expression`|`{"type":"eq","term":"id","value":1}`|
+### Task Serialization
+
+There could be different implementations of tasks,
+e.g., `BaseFileScanTask` and `StaticDataTask` in Java.
+A enum `task-type` field is needed to distinguish different task types.
+
+| Metadata field | JSON representation | Example |
+|-----------------|---------------------|------------------------------------------------------------------------------------------------|
+| **`task-type`** | `JSON string` | `file-scan-task`, `data-task`. Absence of this field should be interpreted as `file-scan-task` |
+
+`file-scan-task` represents a scan task with a data file and maybe delete files.
+It is serialized according to the following table (in addition to the `task-type` field).
+
+| Metadata field |JSON representation|Example|
+|------------------------|--- |--- |
+| **`schema`** |`JSON object`|`See above, read schemas instead`|
Review Comment:
I just realized that I'm not clear on what the example `See above, read schemas instead` is trying to tell the reader. Should this maybe say `See "schemas" section above`?
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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
Re: [PR] Core: add a new task-type field to task JSON serialization. add data task JSON serialization imp. [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #9728:
URL: https://github.com/apache/iceberg/pull/9728#discussion_r1576012052
##########
core/src/test/java/org/apache/iceberg/TestDataTaskParser.java:
##########
@@ -0,0 +1,249 @@
+/*
+ * 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;
+
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.io.StringWriter;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.util.JsonUtil;
+import org.assertj.core.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class TestDataTaskParser {
+ @Test
+ public void nullCheck() throws Exception {
+ StringWriter writer = new StringWriter();
+ JsonGenerator generator = JsonUtil.factory().createGenerator(writer);
+
+ Assertions.assertThatThrownBy(() -> DataTaskParser.toJson(null, generator))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("Invalid data task: null");
+
+ Assertions.assertThatThrownBy(
+ () -> DataTaskParser.toJson((StaticDataTask) createDataTask(), null))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("Invalid JSON generator: null");
+
+ Assertions.assertThatThrownBy(() -> DataTaskParser.fromJson(null))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("Invalid JSON node for data task: null");
+ }
+
+ @Test
+ public void invalidJsonNode() throws Exception {
+ String jsonStr = "{\"str\":\"1\", \"arr\":[]}";
+ ObjectMapper mapper = new ObjectMapper();
+ JsonNode rootNode = mapper.reader().readTree(jsonStr);
+
+ Assertions.assertThatThrownBy(() -> DataTaskParser.fromJson(rootNode.get("str")))
Review Comment:
nit: for new tests it's ok to statically import `assertThatThrownBy()` / `assertThat()`
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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
Re: [PR] Spec, Core: add a new task-type field to task JSON serialization. add data task JSON serialization spec and imp. [iceberg]
Posted by "jackye1995 (via GitHub)" <gi...@apache.org>.
jackye1995 commented on code in PR #9728:
URL: https://github.com/apache/iceberg/pull/9728#discussion_r1492816128
##########
core/src/main/java/org/apache/iceberg/FileScanTaskParser.java:
##########
@@ -21,116 +21,84 @@
import com.fasterxml.jackson.core.JsonGenerator;
import com.fasterxml.jackson.databind.JsonNode;
import java.io.IOException;
-import org.apache.iceberg.expressions.Expression;
-import org.apache.iceberg.expressions.ExpressionParser;
-import org.apache.iceberg.expressions.Expressions;
-import org.apache.iceberg.expressions.ResidualEvaluator;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
-import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.base.Strings;
import org.apache.iceberg.util.JsonUtil;
public class FileScanTaskParser {
- private static final String SCHEMA = "schema";
- private static final String SPEC = "spec";
- private static final String DATA_FILE = "data-file";
- private static final String START = "start";
- private static final String LENGTH = "length";
- private static final String DELETE_FILES = "delete-files";
- private static final String RESIDUAL = "residual-filter";
+ private static final String TASK_TYPE = "task-type";
+
+ private enum TaskType {
Review Comment:
I feel the Java class names are starting to get confusing, even at the interface level. Although `DataTask` extends `FileScanTask`, there is really no file that the `DataTask` is scanning. So although `StaticDataTask` and `BaseFileScanTask` both implements `FileScanTask`, they are not really conceptually similar. The actual relationship is more like `StaticDataTask` implements `DataTask`, `BaseFileScanTask` implements `FileScanTask`, but `DataTask` and `FileScanTask` have many things in common so we made `DataTask` extends `FileScanTask`.
To resolve this confusing situation and potentially accommodate other types of scan task, I am thinking if we should go one more layer above, so we keep the file scan task spec as is, and on top of that, have a serialization spec for **scan task**, which has types like `file-scan-task`, `data-task`. And we then have `ScanTaskParser` that delegates to the existing `FileScanTaskParser` or the `DataTaskParser`. What do you think?
##########
format/spec.md:
##########
@@ -1237,17 +1237,36 @@ Content file (data or delete) is serialized as a JSON object according to the fo
| **`equality-ids`** |`JSON list of int: Field ids used to determine row equality in equality delete files`|`[1]`|
| **`sort-order-id`** |`JSON int`|`1`|
-### File Scan Task Serialization
-
-File scan task is serialized as a JSON object according to the following table.
-
-| Metadata field |JSON representation|Example|
-|--------------------------|--- |--- |
-| **`schema`** |`JSON object`|`See above, read schemas instead`|
-| **`spec`** |`JSON object`|`See above, read partition specs instead`|
-| **`data-file`** |`JSON object`|`See above, read content file instead`|
-| **`delete-files`** |`JSON list of objects`|`See above, read content file instead`|
-| **`residual-filter`** |`JSON object: residual filter expression`|`{"type":"eq","term":"id","value":1}`|
+### Task Serialization
+
+There are different task implementations, e.g. `BaseFileScanTask` and `StaticDataTask` in Java.
+A `task-type` field is needed to distinguish different task types.
+
+| Metadata field | JSON representation | Example |
+|-----------------|---------------------|------------------------------------------------------------------------------------------------|
+| **`task-type`** | `JSON string` | `file-scan-task`, `data-task`. Absence of this field should be interpreted as `file-scan-task` |
+
+`file-scan-task` represents a scan task with a data file and maybe delete files.
Review Comment:
nit: "and maybe delete files" -> "and optional delete files that should be applied to the data file"
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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
Re: [PR] Spec, Core: add a new task-type field to task JSON serialization. add data task JSON serialization spec and imp. [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #9728:
URL: https://github.com/apache/iceberg/pull/9728#discussion_r1493071273
##########
core/src/main/java/org/apache/iceberg/FileScanTaskParser.java:
##########
@@ -21,116 +21,84 @@
import com.fasterxml.jackson.core.JsonGenerator;
import com.fasterxml.jackson.databind.JsonNode;
import java.io.IOException;
-import org.apache.iceberg.expressions.Expression;
-import org.apache.iceberg.expressions.ExpressionParser;
-import org.apache.iceberg.expressions.Expressions;
-import org.apache.iceberg.expressions.ResidualEvaluator;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
-import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.base.Strings;
import org.apache.iceberg.util.JsonUtil;
public class FileScanTaskParser {
- private static final String SCHEMA = "schema";
- private static final String SPEC = "spec";
- private static final String DATA_FILE = "data-file";
- private static final String START = "start";
- private static final String LENGTH = "length";
- private static final String DELETE_FILES = "delete-files";
- private static final String RESIDUAL = "residual-filter";
+ private static final String TASK_TYPE = "task-type";
+
+ private enum TaskType {
Review Comment:
class name refactoring is done.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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
Re: [PR] Core: add a new task-type field to task JSON serialization. add data task JSON serialization imp. [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #9728:
URL: https://github.com/apache/iceberg/pull/9728#discussion_r1576019460
##########
core/src/main/java/org/apache/iceberg/FileScanTaskParser.java:
##########
@@ -40,16 +40,39 @@ public class FileScanTaskParser {
private FileScanTaskParser() {}
+ /**
+ * Serialize file scan task to JSON string
+ *
+ * @deprecated Use {@link ScanTaskParser#toJson(FileScanTask)} instead; will be removed in 1.7.0
+ */
+ @Deprecated
public static String toJson(FileScanTask fileScanTask) {
- return JsonUtil.generate(
- generator -> FileScanTaskParser.toJson(fileScanTask, generator), false);
+ Preconditions.checkArgument(fileScanTask != null, "Invalid file scan task: null");
+ return JsonUtil.generate(generator -> toJsonWrapper(fileScanTask, generator), false);
Review Comment:
I don't think we need this extra method as we can just inline this:
```
@Deprecated
public static String toJson(FileScanTask fileScanTask) {
Preconditions.checkArgument(fileScanTask != null, "Invalid file scan task: null");
return JsonUtil.generate(
generator -> {
generator.writeStartObject();
toJson(fileScanTask, generator);
generator.writeEndObject();
},
false);
}
```
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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
Re: [PR] Core: add a new task-type field to task JSON serialization. add data task JSON serialization imp. [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #9728:
URL: https://github.com/apache/iceberg/pull/9728#discussion_r1526494074
##########
core/src/main/java/org/apache/iceberg/ScanTaskParser.java:
##########
@@ -0,0 +1,104 @@
+/*
+ * 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;
+
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.databind.JsonNode;
+import java.io.IOException;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.base.Strings;
+import org.apache.iceberg.util.JsonUtil;
+
+public class ScanTaskParser {
+ private static final String TASK_TYPE = "task-type";
+
+ private enum TaskType {
+ FILE_SCAN_TASK("file-scan-task"),
+ DATA_TASK("data-task");
+
+ private final String value;
+
+ TaskType(String value) {
+ this.value = value;
+ }
+
+ public static TaskType fromValue(String value) {
+ Preconditions.checkArgument(value != null, "Task type is null");
+ if (FILE_SCAN_TASK.value().equalsIgnoreCase(value)) {
+ return FILE_SCAN_TASK;
+ } else if (DATA_TASK.value().equalsIgnoreCase(value)) {
+ return DATA_TASK;
+ } else {
+ throw new IllegalArgumentException("Unknown task type: " + value);
+ }
+ }
+
+ public String value() {
+ return value;
+ }
+ }
+
+ private ScanTaskParser() {}
+
+ public static String toJson(FileScanTask fileScanTask) {
+ Preconditions.checkArgument(fileScanTask != null, "Invalid file scan task: null");
+ return JsonUtil.generate(generator -> toJson(fileScanTask, generator), false);
+ }
+
+ public static FileScanTask fromJson(String json, boolean caseSensitive) {
+ Preconditions.checkArgument(json != null, "Invalid JSON string for file scan task: null");
+ return JsonUtil.parse(json, node -> fromJson(node, caseSensitive));
+ }
+
+ private static void toJson(FileScanTask fileScanTask, JsonGenerator generator)
+ throws IOException {
+ generator.writeStartObject();
+
+ if (fileScanTask instanceof StaticDataTask) {
+ generator.writeStringField(TASK_TYPE, TaskType.DATA_TASK.value());
+ DataTaskParser.toJson((StaticDataTask) fileScanTask, generator);
+ } else if (fileScanTask instanceof BaseFileScanTask
+ || fileScanTask instanceof BaseFileScanTask.SplitScanTask) {
+ generator.writeStringField(TASK_TYPE, TaskType.FILE_SCAN_TASK.value());
+ FileScanTaskParser.toJson(fileScanTask, generator);
+ } else {
+ throw new UnsupportedOperationException(
+ "Unsupported task type: " + fileScanTask.getClass().getCanonicalName());
+ }
+
+ generator.writeEndObject();
+ }
+
+ private static FileScanTask fromJson(JsonNode jsonNode, boolean caseSensitive) {
+ TaskType taskType = TaskType.FILE_SCAN_TASK;
+ String taskTypeStr = JsonUtil.getStringOrNull(TASK_TYPE, jsonNode);
+ if (!Strings.isNullOrEmpty(taskTypeStr)) {
Review Comment:
sounds good. will also need to the empty string check in the `fromTypeName` method
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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
Re: [PR] Core: add a new task-type field to task JSON serialization. add data task JSON serialization imp. [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #9728:
URL: https://github.com/apache/iceberg/pull/9728#discussion_r1526489610
##########
core/src/main/java/org/apache/iceberg/ScanTaskParser.java:
##########
@@ -0,0 +1,104 @@
+/*
+ * 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;
+
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.databind.JsonNode;
+import java.io.IOException;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.base.Strings;
+import org.apache.iceberg.util.JsonUtil;
+
+public class ScanTaskParser {
+ private static final String TASK_TYPE = "task-type";
+
+ private enum TaskType {
+ FILE_SCAN_TASK("file-scan-task"),
+ DATA_TASK("data-task");
+
+ private final String value;
+
+ TaskType(String value) {
+ this.value = value;
+ }
+
+ public static TaskType fromValue(String value) {
+ Preconditions.checkArgument(value != null, "Task type is null");
+ if (FILE_SCAN_TASK.value().equalsIgnoreCase(value)) {
+ return FILE_SCAN_TASK;
+ } else if (DATA_TASK.value().equalsIgnoreCase(value)) {
+ return DATA_TASK;
+ } else {
+ throw new IllegalArgumentException("Unknown task type: " + value);
Review Comment:
I checked out PR #7145 . I am not sure it has the same applicability here. report metric may be considered `optional` and hence return an unknown report metrics object might be desirable there. but here if a scan task is unknown and can't be parsed properly, we should fail explicitly.
##########
core/src/main/java/org/apache/iceberg/ScanTaskParser.java:
##########
@@ -0,0 +1,104 @@
+/*
+ * 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;
+
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.databind.JsonNode;
+import java.io.IOException;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.base.Strings;
+import org.apache.iceberg.util.JsonUtil;
+
+public class ScanTaskParser {
+ private static final String TASK_TYPE = "task-type";
+
+ private enum TaskType {
+ FILE_SCAN_TASK("file-scan-task"),
Review Comment:
see my reply in the comment below
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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
Re: [PR] Core: add a new task-type field to task JSON serialization. add data task JSON serialization imp. [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #9728:
URL: https://github.com/apache/iceberg/pull/9728#discussion_r1586466491
##########
core/src/test/java/org/apache/iceberg/TestDataTaskParser.java:
##########
@@ -0,0 +1,249 @@
+/*
+ * 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;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.io.StringWriter;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.util.JsonUtil;
+import org.junit.jupiter.api.Test;
+
+public class TestDataTaskParser {
+ @Test
+ public void nullCheck() throws Exception {
+ StringWriter writer = new StringWriter();
+ JsonGenerator generator = JsonUtil.factory().createGenerator(writer);
+
+ assertThatThrownBy(() -> DataTaskParser.toJson(null, generator))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("Invalid data task: null");
+
+ assertThatThrownBy(() -> DataTaskParser.toJson((StaticDataTask) createDataTask(), null))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("Invalid JSON generator: null");
+
+ assertThatThrownBy(() -> DataTaskParser.fromJson(null))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("Invalid JSON node for data task: null");
+ }
+
+ @Test
+ public void invalidJsonNode() throws Exception {
+ String jsonStr = "{\"str\":\"1\", \"arr\":[]}";
+ ObjectMapper mapper = new ObjectMapper();
+ JsonNode rootNode = mapper.reader().readTree(jsonStr);
+
+ assertThatThrownBy(() -> DataTaskParser.fromJson(rootNode.get("str")))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Invalid JSON node for data task: non-object ");
+
+ assertThatThrownBy(() -> DataTaskParser.fromJson(rootNode.get("arr")))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Invalid JSON node for data task: non-object ");
+ }
+
+ @Test
+ public void missingFields() throws Exception {
+ ObjectMapper mapper = new ObjectMapper();
+
+ String missingSchemaStr = "{}";
+ JsonNode missingSchemaNode = mapper.reader().readTree(missingSchemaStr);
+ assertThatThrownBy(() -> DataTaskParser.fromJson(missingSchemaNode))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Cannot parse missing field: schema");
+
+ String missingProjectionStr =
+ "{"
+ + "\"schema\":{\"type\":\"struct\",\"schema-id\":0,"
+ + "\"fields\":[{\"id\":1,\"name\":\"committed_at\",\"required\":true,\"type\":\"timestamptz\"}]}"
+ + "}";
+ JsonNode missingProjectionNode = mapper.reader().readTree(missingProjectionStr);
+ assertThatThrownBy(() -> DataTaskParser.fromJson(missingProjectionNode))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Cannot parse missing field: projection");
+
+ String missingMetadataFileStr =
+ "{"
+ + "\"schema\":{\"type\":\"struct\",\"schema-id\":0,"
+ + "\"fields\":[{\"id\":1,\"name\":\"committed_at\",\"required\":true,\"type\":\"timestamptz\"}]},"
+ + "\"projection\":{\"type\":\"struct\",\"schema-id\":0,"
+ + "\"fields\":[{\"id\":1,\"name\":\"committed_at\",\"required\":true,\"type\":\"timestamptz\"}]}"
+ + "}";
+ JsonNode missingMetadataFileNode = mapper.reader().readTree(missingMetadataFileStr);
+ assertThatThrownBy(() -> DataTaskParser.fromJson(missingMetadataFileNode))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Cannot parse missing field: metadata-file");
+
+ String missingTableRowsStr =
+ "{\"task-type\":\"data-task\","
+ + "\"schema\":{\"type\":\"struct\",\"schema-id\":0,"
+ + "\"fields\":[{\"id\":1,\"name\":\"committed_at\",\"required\":true,\"type\":\"timestamptz\"},"
+ + "{\"id\":2,\"name\":\"snapshot_id\",\"required\":true,\"type\":\"long\"},"
+ + "{\"id\":3,\"name\":\"parent_id\",\"required\":false,\"type\":\"long\"},"
+ + "{\"id\":4,\"name\":\"operation\",\"required\":false,\"type\":\"string\"},"
+ + "{\"id\":5,\"name\":\"manifest_list\",\"required\":false,\"type\":\"string\"},"
+ + "{\"id\":6,\"name\":\"summary\",\"required\":false,\"type\":{\"type\":\"map\","
+ + "\"key-id\":7,\"key\":\"string\",\"value-id\":8,"
+ + "\"value\":\"string\",\"value-required\":true}}]},"
+ + "\"projection\":{\"type\":\"struct\",\"schema-id\":0,"
+ + "\"fields\":[{\"id\":1,\"name\":\"committed_at\",\"required\":true,\"type\":\"timestamptz\"},"
+ + "{\"id\":2,\"name\":\"snapshot_id\",\"required\":true,\"type\":\"long\"},"
+ + "{\"id\":3,\"name\":\"parent_id\",\"required\":false,\"type\":\"long\"},"
+ + "{\"id\":4,\"name\":\"operation\",\"required\":false,\"type\":\"string\"},"
+ + "{\"id\":5,\"name\":\"manifest_list\",\"required\":false,\"type\":\"string\"},"
+ + "{\"id\":6,\"name\":\"summary\",\"required\":false,\"type\":{\"type\":\"map\","
+ + "\"key-id\":7,\"key\":\"string\",\"value-id\":8,"
+ + "\"value\":\"string\",\"value-required\":true}}]},"
+ + "\"metadata-file\":{\"spec-id\":0,\"content\":\"DATA\","
+ + "\"file-path\":\"/tmp/metadata2.json\","
+ + "\"file-format\":\"METADATA\",\"partition\":{},"
+ + "\"file-size-in-bytes\":0,\"record-count\":2,\"sort-order-id\":0}"
+ + "}";
+ JsonNode missingTableRowsNode = mapper.reader().readTree(missingTableRowsStr);
+ assertThatThrownBy(() -> DataTaskParser.fromJson(missingTableRowsNode))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Cannot parse missing field: rows");
+ }
+
+ @Test
+ public void roundTripSerde() {
+ StaticDataTask dataTask = (StaticDataTask) createDataTask();
+ String jsonStr = ScanTaskParser.toJson(dataTask);
+ assertThat(jsonStr).isEqualTo(snapshotsDataTaskJson());
+ StaticDataTask deserializedTask = (StaticDataTask) ScanTaskParser.fromJson(jsonStr, true);
+ assertDataTaskEquals(dataTask, deserializedTask);
+ }
+
+ private DataTask createDataTask() {
+ Map<String, String> summary1 =
+ ImmutableMap.of(
+ "added-data-files", "1",
+ "added-records", "1",
+ "added-files-size", "10",
+ "changed-partition-count", "1",
+ "total-records", "1",
+ "total-files-size", "10",
+ "total-data-files", "1",
+ "total-delete-files", "0",
+ "total-position-deletes", "0",
+ "total-equality-deletes", "0");
+
+ Map<String, String> summary2 =
+ ImmutableMap.of(
+ "added-data-files", "1",
+ "added-records", "1",
+ "added-files-size", "10",
+ "changed-partition-count", "1",
+ "total-records", "2",
+ "total-files-size", "20",
+ "total-data-files", "2",
+ "total-delete-files", "0",
+ "total-position-deletes", "0",
+ "total-equality-deletes", "0");
+
+ List<Snapshot> snapshots =
+ Arrays.asList(
+ new BaseSnapshot(
+ 1L, 1L, null, 1234567890000L, "append", summary1, 1, "file:/tmp/manifest1.avro"),
+ new BaseSnapshot(
+ 2L, 2L, 1L, 9876543210000L, "append", summary2, 1, "file:/tmp/manifest2.avro"));
+
+ return StaticDataTask.of(
+ Files.localInput("file:/tmp/metadata2.json"),
+ SnapshotsTable.SNAPSHOT_SCHEMA,
+ SnapshotsTable.SNAPSHOT_SCHEMA,
+ snapshots,
+ SnapshotsTable::snapshotToRow);
+ }
+
+ private String snapshotsDataTaskJson() {
+ return "{\"task-type\":\"data-task\","
+ + "\"schema\":{\"type\":\"struct\",\"schema-id\":0,"
+ + "\"fields\":[{\"id\":1,\"name\":\"committed_at\",\"required\":true,\"type\":\"timestamptz\"},"
+ + "{\"id\":2,\"name\":\"snapshot_id\",\"required\":true,\"type\":\"long\"},"
+ + "{\"id\":3,\"name\":\"parent_id\",\"required\":false,\"type\":\"long\"},"
+ + "{\"id\":4,\"name\":\"operation\",\"required\":false,\"type\":\"string\"},"
+ + "{\"id\":5,\"name\":\"manifest_list\",\"required\":false,\"type\":\"string\"},"
+ + "{\"id\":6,\"name\":\"summary\",\"required\":false,\"type\":{\"type\":\"map\","
+ + "\"key-id\":7,\"key\":\"string\",\"value-id\":8,"
+ + "\"value\":\"string\",\"value-required\":true}}]},"
+ + "\"projection\":{\"type\":\"struct\",\"schema-id\":0,"
+ + "\"fields\":[{\"id\":1,\"name\":\"committed_at\",\"required\":true,\"type\":\"timestamptz\"},"
+ + "{\"id\":2,\"name\":\"snapshot_id\",\"required\":true,\"type\":\"long\"},"
+ + "{\"id\":3,\"name\":\"parent_id\",\"required\":false,\"type\":\"long\"},"
+ + "{\"id\":4,\"name\":\"operation\",\"required\":false,\"type\":\"string\"},"
+ + "{\"id\":5,\"name\":\"manifest_list\",\"required\":false,\"type\":\"string\"},"
+ + "{\"id\":6,\"name\":\"summary\",\"required\":false,\"type\":{\"type\":\"map\","
+ + "\"key-id\":7,\"key\":\"string\",\"value-id\":8,"
+ + "\"value\":\"string\",\"value-required\":true}}]},"
+ + "\"metadata-file\":{\"spec-id\":0,\"content\":\"DATA\","
+ + "\"file-path\":\"/tmp/metadata2.json\","
+ + "\"file-format\":\"METADATA\",\"partition\":{},"
+ + "\"file-size-in-bytes\":0,\"record-count\":2,\"sort-order-id\":0},"
+ + "\"rows\":[{\"1\":\"2009-02-13T23:31:30+00:00\",\"2\":1,\"4\":\"append\","
+ + "\"5\":\"file:/tmp/manifest1.avro\","
+ + "\"6\":{\"keys\":[\"added-data-files\",\"added-records\",\"added-files-size\",\"changed-partition-count\","
+ + "\"total-records\",\"total-files-size\",\"total-data-files\",\"total-delete-files\","
+ + "\"total-position-deletes\",\"total-equality-deletes\"],"
+ + "\"values\":[\"1\",\"1\",\"10\",\"1\",\"1\",\"10\",\"1\",\"0\",\"0\",\"0\"]}},"
+ + "{\"1\":\"2282-12-22T20:13:30+00:00\",\"2\":2,\"3\":1,\"4\":\"append\","
+ + "\"5\":\"file:/tmp/manifest2.avro\","
+ + "\"6\":{\"keys\":[\"added-data-files\",\"added-records\",\"added-files-size\",\"changed-partition-count\","
+ + "\"total-records\",\"total-files-size\",\"total-data-files\",\"total-delete-files\","
+ + "\"total-position-deletes\",\"total-equality-deletes\"],"
+ + "\"values\":[\"1\",\"1\",\"10\",\"1\",\"2\",\"20\",\"2\",\"0\",\"0\",\"0\"]}}]}";
+ }
+
+ private void assertDataTaskEquals(StaticDataTask expected, StaticDataTask actual) {
+ assertThat(expected.schema().asStruct())
Review Comment:
thanks for catching the mistake
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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
Re: [PR] Core: add a new task-type field to task JSON serialization. add data task JSON serialization imp. [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #9728:
URL: https://github.com/apache/iceberg/pull/9728#discussion_r1582996766
##########
core/src/main/java/org/apache/iceberg/SnapshotsTable.java:
##########
@@ -27,7 +28,8 @@
* <p>This does not include snapshots that have been expired using {@link ExpireSnapshots}.
*/
public class SnapshotsTable extends BaseMetadataTable {
- private static final Schema SNAPSHOT_SCHEMA =
+ @VisibleForTesting
+ static final Schema SNAPSHOT_SCHEMA =
Review Comment:
I think having a custom schema for the test makes sense 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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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
Re: [PR] Spec, Core: add a new task-type field to task JSON serialization. add data task JSON serialization spec and imp. [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #9728:
URL: https://github.com/apache/iceberg/pull/9728#discussion_r1492723658
##########
core/src/main/java/org/apache/iceberg/BaseFileScanTaskParser.java:
##########
@@ -0,0 +1,122 @@
+/*
+ * 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;
+
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.databind.JsonNode;
+import java.io.IOException;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.ExpressionParser;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.expressions.ResidualEvaluator;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.util.JsonUtil;
+
+class BaseFileScanTaskParser {
Review Comment:
code here is moved from the `FileScanTaskParser` who becomes a facade/dispatcher class now
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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
Re: [PR] Spec, Core: add a new task-type field to task JSON serialization. add data task JSON serialization spec and imp. [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #9728:
URL: https://github.com/apache/iceberg/pull/9728#discussion_r1493074229
##########
format/spec.md:
##########
@@ -1237,17 +1237,36 @@ Content file (data or delete) is serialized as a JSON object according to the fo
| **`equality-ids`** |`JSON list of int: Field ids used to determine row equality in equality delete files`|`[1]`|
| **`sort-order-id`** |`JSON int`|`1`|
-### File Scan Task Serialization
-
-File scan task is serialized as a JSON object according to the following table.
-
-| Metadata field |JSON representation|Example|
-|--------------------------|--- |--- |
-| **`schema`** |`JSON object`|`See above, read schemas instead`|
-| **`spec`** |`JSON object`|`See above, read partition specs instead`|
-| **`data-file`** |`JSON object`|`See above, read content file instead`|
-| **`delete-files`** |`JSON list of objects`|`See above, read content file instead`|
-| **`residual-filter`** |`JSON object: residual filter expression`|`{"type":"eq","term":"id","value":1}`|
+### Task Serialization
Review Comment:
Flink source checkpoints pending splits of `FileScanTask`. The standardization of JSON serialization scan task is used by both REST OpenAPI and Flink (checkpoint and job manager -> task manager split assignment). I would imagine if Spark streaming checkpoint pending splits (scan task), it would probably also prefer to use JSON serialization (than Java serialization).
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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
Re: [PR] Spec, Core: add a new task-type field to task JSON serialization. add data task JSON serialization spec and imp. [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #9728:
URL: https://github.com/apache/iceberg/pull/9728#discussion_r1493034865
##########
core/src/main/java/org/apache/iceberg/FileScanTaskParser.java:
##########
@@ -21,116 +21,84 @@
import com.fasterxml.jackson.core.JsonGenerator;
import com.fasterxml.jackson.databind.JsonNode;
import java.io.IOException;
-import org.apache.iceberg.expressions.Expression;
-import org.apache.iceberg.expressions.ExpressionParser;
-import org.apache.iceberg.expressions.Expressions;
-import org.apache.iceberg.expressions.ResidualEvaluator;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
-import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.base.Strings;
import org.apache.iceberg.util.JsonUtil;
public class FileScanTaskParser {
- private static final String SCHEMA = "schema";
- private static final String SPEC = "spec";
- private static final String DATA_FILE = "data-file";
- private static final String START = "start";
- private static final String LENGTH = "length";
- private static final String DELETE_FILES = "delete-files";
- private static final String RESIDUAL = "residual-filter";
+ private static final String TASK_TYPE = "task-type";
+
+ private enum TaskType {
Review Comment:
@jackye1995 I agree that `ScanTaskParser` is more accurate here. The only problem is that `FileScanTaskParser` is a public class in core module. Technically, we are breaking compatibility.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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
Re: [PR] Spec, Core: add a new task-type field to task JSON serialization. add data task JSON serialization spec and imp. [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #9728:
URL: https://github.com/apache/iceberg/pull/9728#discussion_r1493034865
##########
core/src/main/java/org/apache/iceberg/FileScanTaskParser.java:
##########
@@ -21,116 +21,84 @@
import com.fasterxml.jackson.core.JsonGenerator;
import com.fasterxml.jackson.databind.JsonNode;
import java.io.IOException;
-import org.apache.iceberg.expressions.Expression;
-import org.apache.iceberg.expressions.ExpressionParser;
-import org.apache.iceberg.expressions.Expressions;
-import org.apache.iceberg.expressions.ResidualEvaluator;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
-import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.base.Strings;
import org.apache.iceberg.util.JsonUtil;
public class FileScanTaskParser {
- private static final String SCHEMA = "schema";
- private static final String SPEC = "spec";
- private static final String DATA_FILE = "data-file";
- private static final String START = "start";
- private static final String LENGTH = "length";
- private static final String DELETE_FILES = "delete-files";
- private static final String RESIDUAL = "residual-filter";
+ private static final String TASK_TYPE = "task-type";
+
+ private enum TaskType {
Review Comment:
@jackye1995 I agree that `ScanTaskParser` is more accurate for the facade/dispatcher class here. The only problem is that `FileScanTaskParser` is a public class in core module. Technically, we are breaking compatibility.
maybe we can keep `FileScanTaskParser` and mark it as deprecated. it can just extend from the new `ScanTaskParser` to avoid code duplication.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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
Re: [PR] Core: add a new task-type field to task JSON serialization. add data task JSON serialization imp. [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #9728:
URL: https://github.com/apache/iceberg/pull/9728#discussion_r1526510194
##########
core/src/main/java/org/apache/iceberg/DataTaskParser.java:
##########
@@ -0,0 +1,80 @@
+/*
+ * 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;
+
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.databind.JsonNode;
+import java.io.IOException;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.util.JsonUtil;
+
+class DataTaskParser {
+ private static final String SCHEMA = "schema";
+ private static final String PROJECTED_SCHEMA = "projection";
+ private static final String METADATA_FILE = "metadata-file";
+ private static final String ROWS = "rows";
+
+ private DataTaskParser() {}
+
+ static void toJson(StaticDataTask dataTask, JsonGenerator generator) throws IOException {
+ Preconditions.checkArgument(dataTask != null, "Invalid data task: null");
+ Preconditions.checkArgument(generator != null, "Invalid JSON generator: null");
+
+ generator.writeFieldName(SCHEMA);
+ SchemaParser.toJson(dataTask.schema(), generator);
+
+ generator.writeFieldName(PROJECTED_SCHEMA);
+ SchemaParser.toJson(dataTask.projectedSchema(), generator);
+
+ generator.writeFieldName(METADATA_FILE);
+ ContentFileParser.toJson(dataTask.metadataFile(), PartitionSpec.unpartitioned(), generator);
+
+ generator.writeArrayFieldStart(ROWS);
+ for (StructLike row : dataTask.tableRows()) {
Review Comment:
I suppose it can be empty. that is fine as parser will just generate an empty array in JSON. it can't be null. we can add a null check here if that is desirable. but the StaticDataTask constructor/factory ensured non-null.
```
class StaticDataTask implements DataTask {
static <T> DataTask of(
InputFile metadata,
Schema tableSchema,
Schema projectedSchema,
Iterable<T> values,
Function<T, Row> transform) {
return new StaticDataTask(
metadata,
tableSchema,
projectedSchema,
Lists.newArrayList(Iterables.transform(values, transform::apply)).toArray(new Row[0]));
}
```
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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
Re: [PR] Core: add a new task-type field to task JSON serialization. add data task JSON serialization implementation. [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #9728:
URL: https://github.com/apache/iceberg/pull/9728#discussion_r1647487236
##########
core/src/main/java/org/apache/iceberg/DataTaskParser.java:
##########
@@ -0,0 +1,81 @@
+/*
+ * 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;
+
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.databind.JsonNode;
+import java.io.IOException;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.util.JsonUtil;
+
+class DataTaskParser {
+ private static final String SCHEMA = "schema";
+ private static final String PROJECTED_SCHEMA = "projection";
+ private static final String METADATA_FILE = "metadata-file";
+ private static final String ROWS = "rows";
+
+ private DataTaskParser() {}
Review Comment:
nit: Why is this not `StaticDataTaskParser`?
For me the natural thing would be to have a 1-on-1 connection between the 2
##########
core/src/main/java/org/apache/iceberg/ScanTaskParser.java:
##########
@@ -0,0 +1,105 @@
+/*
+ * 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;
+
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.databind.JsonNode;
+import java.io.IOException;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.base.Strings;
+import org.apache.iceberg.util.JsonUtil;
+
+public class ScanTaskParser {
+ private static final String TASK_TYPE = "task-type";
+
+ private enum TaskType {
+ FILE_SCAN_TASK("file-scan-task"),
+ DATA_TASK("data-task");
Review Comment:
nit: Maybe `STATIC_DATA_TASK`?
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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
Re: [PR] Spec, Core: add a new task-type field to task JSON serialization. add data task JSON serialization spec and imp. [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #9728:
URL: https://github.com/apache/iceberg/pull/9728#discussion_r1493050495
##########
core/src/main/java/org/apache/iceberg/FileScanTaskParser.java:
##########
@@ -21,116 +21,84 @@
import com.fasterxml.jackson.core.JsonGenerator;
import com.fasterxml.jackson.databind.JsonNode;
import java.io.IOException;
-import org.apache.iceberg.expressions.Expression;
-import org.apache.iceberg.expressions.ExpressionParser;
-import org.apache.iceberg.expressions.Expressions;
-import org.apache.iceberg.expressions.ResidualEvaluator;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
-import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.base.Strings;
import org.apache.iceberg.util.JsonUtil;
public class FileScanTaskParser {
- private static final String SCHEMA = "schema";
- private static final String SPEC = "spec";
- private static final String DATA_FILE = "data-file";
- private static final String START = "start";
- private static final String LENGTH = "length";
- private static final String DELETE_FILES = "delete-files";
- private static final String RESIDUAL = "residual-filter";
+ private static final String TASK_TYPE = "task-type";
+
+ private enum TaskType {
Review Comment:
actually, we can just deprecate the public methods from `FileScanTaskParser` and still keep the file scan task impl 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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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
Re: [PR] Spec: add task-type field to JSON serialization of file scan task. add JSON serialization for StaticDataTask. [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #9728:
URL: https://github.com/apache/iceberg/pull/9728#discussion_r1490214139
##########
format/spec.md:
##########
@@ -1239,15 +1239,34 @@ Content file (data or delete) is serialized as a JSON object according to the fo
### File Scan Task Serialization
-File scan task is serialized as a JSON object according to the following table.
-
-| Metadata field |JSON representation|Example|
-|--------------------------|--- |--- |
-| **`schema`** |`JSON object`|`See above, read schemas instead`|
-| **`spec`** |`JSON object`|`See above, read partition specs instead`|
-| **`data-file`** |`JSON object`|`See above, read content file instead`|
-| **`delete-files`** |`JSON list of objects`|`See above, read content file instead`|
-| **`residual-filter`** |`JSON object: residual filter expression`|`{"type":"eq","term":"id","value":1}`|
+There could be different implementations of file scan task,
+e.g., `BaseFileScanTask` and `StaticDataTask` in Java.
+A enum `task-type` field is needed to distinguish different task types.
+
+| Metadata field | JSON representation | Example |
+|-----------------|---------------------|---------------------------------------------------------------------------------------|
+| **`task-type`** | `JSON string` | `base-file-task`, `static-data-task`. Absence of this field should be interpreted as `base-file-task` |
Review Comment:
the suggestions make sense. updated the naming.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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
Re: [PR] Spec, Core: add a new task-type field to task JSON serialization. add data task JSON serialization spec and imp. [iceberg]
Posted by "emkornfield (via GitHub)" <gi...@apache.org>.
emkornfield commented on code in PR #9728:
URL: https://github.com/apache/iceberg/pull/9728#discussion_r1494889470
##########
format/spec.md:
##########
@@ -1237,17 +1237,36 @@ Content file (data or delete) is serialized as a JSON object according to the fo
| **`equality-ids`** |`JSON list of int: Field ids used to determine row equality in equality delete files`|`[1]`|
| **`sort-order-id`** |`JSON int`|`1`|
-### File Scan Task Serialization
-
-File scan task is serialized as a JSON object according to the following table.
-
-| Metadata field |JSON representation|Example|
-|--------------------------|--- |--- |
-| **`schema`** |`JSON object`|`See above, read schemas instead`|
-| **`spec`** |`JSON object`|`See above, read partition specs instead`|
-| **`data-file`** |`JSON object`|`See above, read content file instead`|
-| **`delete-files`** |`JSON list of objects`|`See above, read content file instead`|
-| **`residual-filter`** |`JSON object: residual filter expression`|`{"type":"eq","term":"id","value":1}`|
+### Task Serialization
Review Comment:
IMO, unless we expect Spark to read the Rest checkpoints from Flink (I would guess this isn't the case), then this is really an implementation detail of both engines and doesn't really belong in the specification.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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
Re: [PR] Core: add a new task-type field to task JSON serialization. add data task JSON serialization imp. [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #9728:
URL: https://github.com/apache/iceberg/pull/9728#discussion_r1526583765
##########
core/src/test/java/org/apache/iceberg/TestDataTaskParser.java:
##########
@@ -0,0 +1,147 @@
+/*
+ * 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;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.assertj.core.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class TestDataTaskParser {
+ @Test
+ public void testStaticDataTask() {
+ StaticDataTask dataTask = (StaticDataTask) createDataTask();
+ String jsonStr = ScanTaskParser.toJson(dataTask);
+ Assertions.assertThat(jsonStr).isEqualTo(snapshotsDataTaskJson());
+ StaticDataTask deserializedTask = (StaticDataTask) ScanTaskParser.fromJson(jsonStr, true);
+ assertDataTaskEquals(dataTask, deserializedTask);
+ }
+
+ private DataTask createDataTask() {
+ Map<String, String> summary1 =
+ ImmutableMap.of(
+ "added-data-files", "1",
+ "added-records", "1",
+ "added-files-size", "10",
+ "changed-partition-count", "1",
+ "total-records", "1",
+ "total-files-size", "10",
+ "total-data-files", "1",
+ "total-delete-files", "0",
+ "total-position-deletes", "0",
+ "total-equality-deletes", "0");
+
+ Map<String, String> summary2 =
+ ImmutableMap.of(
+ "added-data-files", "1",
+ "added-records", "1",
+ "added-files-size", "10",
+ "changed-partition-count", "1",
+ "total-records", "2",
+ "total-files-size", "20",
+ "total-data-files", "2",
+ "total-delete-files", "0",
+ "total-position-deletes", "0",
+ "total-equality-deletes", "0");
+
+ List<Snapshot> snapshots =
+ Arrays.asList(
+ new BaseSnapshot(
+ 1L, 1L, null, 1234567890000L, "append", summary1, 1, "file:/tmp/manifest1.avro"),
+ new BaseSnapshot(
+ 2L, 2L, 1L, 9876543210000L, "append", summary2, 1, "file:/tmp/manifest2.avro"));
+
+ return StaticDataTask.of(
+ Files.localInput("file:/tmp/metadata2.json"),
+ SnapshotsTable.SNAPSHOT_SCHEMA,
+ SnapshotsTable.SNAPSHOT_SCHEMA,
+ snapshots,
+ SnapshotsTable::snapshotToRow);
+ }
+
+ private String snapshotsDataTaskJson() {
+ return "{\"task-type\":\"data-task\","
Review Comment:
scan task parser doesn't support pretty format. hence can't have the expected string to be pretty formatted
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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
Re: [PR] Core: add a new task-type field to task JSON serialization. add data task JSON serialization imp. [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #9728:
URL: https://github.com/apache/iceberg/pull/9728#discussion_r1576011200
##########
core/src/test/java/org/apache/iceberg/TestDataTaskParser.java:
##########
@@ -0,0 +1,249 @@
+/*
+ * 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;
+
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.io.StringWriter;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.util.JsonUtil;
+import org.assertj.core.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class TestDataTaskParser {
+ @Test
+ public void nullCheck() throws Exception {
+ StringWriter writer = new StringWriter();
+ JsonGenerator generator = JsonUtil.factory().createGenerator(writer);
+
+ Assertions.assertThatThrownBy(() -> DataTaskParser.toJson(null, generator))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("Invalid data task: null");
+
+ Assertions.assertThatThrownBy(
+ () -> DataTaskParser.toJson((StaticDataTask) createDataTask(), null))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("Invalid JSON generator: null");
+
+ Assertions.assertThatThrownBy(() -> DataTaskParser.fromJson(null))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("Invalid JSON node for data task: null");
+ }
+
+ @Test
+ public void invalidJsonNode() throws Exception {
+ String jsonStr = "{\"str\":\"1\", \"arr\":[]}";
+ ObjectMapper mapper = new ObjectMapper();
+ JsonNode rootNode = mapper.reader().readTree(jsonStr);
+
+ Assertions.assertThatThrownBy(() -> DataTaskParser.fromJson(rootNode.get("str")))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Invalid JSON node for data task: non-object ");
+
+ Assertions.assertThatThrownBy(() -> DataTaskParser.fromJson(rootNode.get("arr")))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Invalid JSON node for data task: non-object ");
+ }
+
+ @Test
+ public void missingFields() throws Exception {
+ ObjectMapper mapper = new ObjectMapper();
+
+ String missingSchemaStr = "{}";
+ JsonNode missingSchemaNode = mapper.reader().readTree(missingSchemaStr);
+ Assertions.assertThatThrownBy(() -> DataTaskParser.fromJson(missingSchemaNode))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Cannot parse missing field: schema");
+
+ String missingProjectionStr =
+ "{"
+ + "\"schema\":{\"type\":\"struct\",\"schema-id\":0,"
+ + "\"fields\":[{\"id\":1,\"name\":\"committed_at\",\"required\":true,\"type\":\"timestamptz\"}]}"
+ + "}";
+ JsonNode missingProjectionNode = mapper.reader().readTree(missingProjectionStr);
+ Assertions.assertThatThrownBy(() -> DataTaskParser.fromJson(missingProjectionNode))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Cannot parse missing field: projection");
+
+ String missingMetadataFileStr =
+ "{"
+ + "\"schema\":{\"type\":\"struct\",\"schema-id\":0,"
+ + "\"fields\":[{\"id\":1,\"name\":\"committed_at\",\"required\":true,\"type\":\"timestamptz\"}]},"
+ + "\"projection\":{\"type\":\"struct\",\"schema-id\":0,"
+ + "\"fields\":[{\"id\":1,\"name\":\"committed_at\",\"required\":true,\"type\":\"timestamptz\"}]}"
+ + "}";
+ JsonNode missingMetadataFileNode = mapper.reader().readTree(missingMetadataFileStr);
+ Assertions.assertThatThrownBy(() -> DataTaskParser.fromJson(missingMetadataFileNode))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Cannot parse missing field: metadata-file");
+
+ String missingTableRowsStr =
+ "{\"task-type\":\"data-task\","
+ + "\"schema\":{\"type\":\"struct\",\"schema-id\":0,"
+ + "\"fields\":[{\"id\":1,\"name\":\"committed_at\",\"required\":true,\"type\":\"timestamptz\"},"
+ + "{\"id\":2,\"name\":\"snapshot_id\",\"required\":true,\"type\":\"long\"},"
+ + "{\"id\":3,\"name\":\"parent_id\",\"required\":false,\"type\":\"long\"},"
+ + "{\"id\":4,\"name\":\"operation\",\"required\":false,\"type\":\"string\"},"
+ + "{\"id\":5,\"name\":\"manifest_list\",\"required\":false,\"type\":\"string\"},"
+ + "{\"id\":6,\"name\":\"summary\",\"required\":false,\"type\":{\"type\":\"map\","
+ + "\"key-id\":7,\"key\":\"string\",\"value-id\":8,"
+ + "\"value\":\"string\",\"value-required\":true}}]},"
+ + "\"projection\":{\"type\":\"struct\",\"schema-id\":0,"
+ + "\"fields\":[{\"id\":1,\"name\":\"committed_at\",\"required\":true,\"type\":\"timestamptz\"},"
+ + "{\"id\":2,\"name\":\"snapshot_id\",\"required\":true,\"type\":\"long\"},"
+ + "{\"id\":3,\"name\":\"parent_id\",\"required\":false,\"type\":\"long\"},"
+ + "{\"id\":4,\"name\":\"operation\",\"required\":false,\"type\":\"string\"},"
+ + "{\"id\":5,\"name\":\"manifest_list\",\"required\":false,\"type\":\"string\"},"
+ + "{\"id\":6,\"name\":\"summary\",\"required\":false,\"type\":{\"type\":\"map\","
+ + "\"key-id\":7,\"key\":\"string\",\"value-id\":8,"
+ + "\"value\":\"string\",\"value-required\":true}}]},"
+ + "\"metadata-file\":{\"spec-id\":0,\"content\":\"DATA\","
+ + "\"file-path\":\"/tmp/metadata2.json\","
+ + "\"file-format\":\"METADATA\",\"partition\":{},"
+ + "\"file-size-in-bytes\":0,\"record-count\":2,\"sort-order-id\":0}"
+ + "}";
+ JsonNode missingTableRowsNode = mapper.reader().readTree(missingTableRowsStr);
+ Assertions.assertThatThrownBy(() -> DataTaskParser.fromJson(missingTableRowsNode))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Cannot parse missing field: rows");
+ }
+
+ @Test
+ public void roundTripSerde() {
+ StaticDataTask dataTask = (StaticDataTask) createDataTask();
+ String jsonStr = ScanTaskParser.toJson(dataTask);
+ Assertions.assertThat(jsonStr).isEqualTo(snapshotsDataTaskJson());
+ StaticDataTask deserializedTask = (StaticDataTask) ScanTaskParser.fromJson(jsonStr, true);
+ assertDataTaskEquals(dataTask, deserializedTask);
+ }
+
+ private DataTask createDataTask() {
+ Map<String, String> summary1 =
+ ImmutableMap.of(
+ "added-data-files", "1",
+ "added-records", "1",
+ "added-files-size", "10",
+ "changed-partition-count", "1",
+ "total-records", "1",
+ "total-files-size", "10",
+ "total-data-files", "1",
+ "total-delete-files", "0",
+ "total-position-deletes", "0",
+ "total-equality-deletes", "0");
+
+ Map<String, String> summary2 =
+ ImmutableMap.of(
+ "added-data-files", "1",
+ "added-records", "1",
+ "added-files-size", "10",
+ "changed-partition-count", "1",
+ "total-records", "2",
+ "total-files-size", "20",
+ "total-data-files", "2",
+ "total-delete-files", "0",
+ "total-position-deletes", "0",
+ "total-equality-deletes", "0");
+
+ List<Snapshot> snapshots =
+ Arrays.asList(
+ new BaseSnapshot(
+ 1L, 1L, null, 1234567890000L, "append", summary1, 1, "file:/tmp/manifest1.avro"),
+ new BaseSnapshot(
+ 2L, 2L, 1L, 9876543210000L, "append", summary2, 1, "file:/tmp/manifest2.avro"));
+
+ return StaticDataTask.of(
+ Files.localInput("file:/tmp/metadata2.json"),
+ SnapshotsTable.SNAPSHOT_SCHEMA,
+ SnapshotsTable.SNAPSHOT_SCHEMA,
+ snapshots,
+ SnapshotsTable::snapshotToRow);
+ }
+
+ private String snapshotsDataTaskJson() {
+ return "{\"task-type\":\"data-task\","
+ + "\"schema\":{\"type\":\"struct\",\"schema-id\":0,"
+ + "\"fields\":[{\"id\":1,\"name\":\"committed_at\",\"required\":true,\"type\":\"timestamptz\"},"
+ + "{\"id\":2,\"name\":\"snapshot_id\",\"required\":true,\"type\":\"long\"},"
+ + "{\"id\":3,\"name\":\"parent_id\",\"required\":false,\"type\":\"long\"},"
+ + "{\"id\":4,\"name\":\"operation\",\"required\":false,\"type\":\"string\"},"
+ + "{\"id\":5,\"name\":\"manifest_list\",\"required\":false,\"type\":\"string\"},"
+ + "{\"id\":6,\"name\":\"summary\",\"required\":false,\"type\":{\"type\":\"map\","
+ + "\"key-id\":7,\"key\":\"string\",\"value-id\":8,"
+ + "\"value\":\"string\",\"value-required\":true}}]},"
+ + "\"projection\":{\"type\":\"struct\",\"schema-id\":0,"
+ + "\"fields\":[{\"id\":1,\"name\":\"committed_at\",\"required\":true,\"type\":\"timestamptz\"},"
+ + "{\"id\":2,\"name\":\"snapshot_id\",\"required\":true,\"type\":\"long\"},"
+ + "{\"id\":3,\"name\":\"parent_id\",\"required\":false,\"type\":\"long\"},"
+ + "{\"id\":4,\"name\":\"operation\",\"required\":false,\"type\":\"string\"},"
+ + "{\"id\":5,\"name\":\"manifest_list\",\"required\":false,\"type\":\"string\"},"
+ + "{\"id\":6,\"name\":\"summary\",\"required\":false,\"type\":{\"type\":\"map\","
+ + "\"key-id\":7,\"key\":\"string\",\"value-id\":8,"
+ + "\"value\":\"string\",\"value-required\":true}}]},"
+ + "\"metadata-file\":{\"spec-id\":0,\"content\":\"DATA\","
+ + "\"file-path\":\"/tmp/metadata2.json\","
+ + "\"file-format\":\"METADATA\",\"partition\":{},"
+ + "\"file-size-in-bytes\":0,\"record-count\":2,\"sort-order-id\":0},"
+ + "\"rows\":[{\"1\":\"2009-02-13T23:31:30+00:00\",\"2\":1,\"4\":\"append\","
+ + "\"5\":\"file:/tmp/manifest1.avro\","
+ + "\"6\":{\"keys\":[\"added-data-files\",\"added-records\",\"added-files-size\",\"changed-partition-count\","
+ + "\"total-records\",\"total-files-size\",\"total-data-files\",\"total-delete-files\","
+ + "\"total-position-deletes\",\"total-equality-deletes\"],"
+ + "\"values\":[\"1\",\"1\",\"10\",\"1\",\"1\",\"10\",\"1\",\"0\",\"0\",\"0\"]}},"
+ + "{\"1\":\"2282-12-22T20:13:30+00:00\",\"2\":2,\"3\":1,\"4\":\"append\","
+ + "\"5\":\"file:/tmp/manifest2.avro\","
+ + "\"6\":{\"keys\":[\"added-data-files\",\"added-records\",\"added-files-size\",\"changed-partition-count\","
+ + "\"total-records\",\"total-files-size\",\"total-data-files\",\"total-delete-files\","
+ + "\"total-position-deletes\",\"total-equality-deletes\"],"
+ + "\"values\":[\"1\",\"1\",\"10\",\"1\",\"2\",\"20\",\"2\",\"0\",\"0\",\"0\"]}}]}";
+ }
+
+ private void assertDataTaskEquals(StaticDataTask expected, StaticDataTask actual) {
+ Assertions.assertThat(expected.schema().asStruct())
+ .isEqualTo(actual.schema().asStruct())
+ .as("Schema should match");
Review Comment:
`.as()` needs to come before `.isEqualTo()`. Same for all the other places 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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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
Re: [PR] Spec, Core: add a new task-type field to task JSON serialization. add data task JSON serialization spec and imp. [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #9728:
URL: https://github.com/apache/iceberg/pull/9728#discussion_r1495072935
##########
format/spec.md:
##########
@@ -1237,17 +1237,36 @@ Content file (data or delete) is serialized as a JSON object according to the fo
| **`equality-ids`** |`JSON list of int: Field ids used to determine row equality in equality delete files`|`[1]`|
| **`sort-order-id`** |`JSON int`|`1`|
-### File Scan Task Serialization
-
-File scan task is serialized as a JSON object according to the following table.
-
-| Metadata field |JSON representation|Example|
-|--------------------------|--- |--- |
-| **`schema`** |`JSON object`|`See above, read schemas instead`|
-| **`spec`** |`JSON object`|`See above, read partition specs instead`|
-| **`data-file`** |`JSON object`|`See above, read content file instead`|
-| **`delete-files`** |`JSON list of objects`|`See above, read content file instead`|
-| **`residual-filter`** |`JSON object: residual filter expression`|`{"type":"eq","term":"id","value":1}`|
+### Task Serialization
Review Comment:
got it. @rdblue should I just remove this section from the table spec? not sure what's the policy of removing an invalid section from spec.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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
Re: [PR] Core: add a new task-type field to task JSON serialization. add data task JSON serialization imp. [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #9728:
URL: https://github.com/apache/iceberg/pull/9728#discussion_r1524824279
##########
core/src/main/java/org/apache/iceberg/ScanTaskParser.java:
##########
@@ -0,0 +1,104 @@
+/*
+ * 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;
+
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.databind.JsonNode;
+import java.io.IOException;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.base.Strings;
+import org.apache.iceberg.util.JsonUtil;
+
+public class ScanTaskParser {
+ private static final String TASK_TYPE = "task-type";
+
+ private enum TaskType {
+ FILE_SCAN_TASK("file-scan-task"),
+ DATA_TASK("data-task");
+
+ private final String value;
+
+ TaskType(String value) {
+ this.value = value;
+ }
+
+ public static TaskType fromValue(String value) {
Review Comment:
the codebase typically names this `fromString()` or `fromName()`. Given that this looks specifically for the task type name, maybe this should be named `fromTypeName()`?
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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
Re: [PR] Core: add a new task-type field to task JSON serialization. add data task JSON serialization imp. [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #9728:
URL: https://github.com/apache/iceberg/pull/9728#discussion_r1526625686
##########
core/src/test/java/org/apache/iceberg/TestFileScanTaskParser.java:
##########
@@ -27,29 +27,41 @@
import org.junit.jupiter.params.provider.ValueSource;
public class TestFileScanTaskParser {
+
@Test
public void testNullArguments() {
- Assertions.assertThatThrownBy(() -> FileScanTaskParser.toJson(null))
+ Assertions.assertThatThrownBy(() -> ScanTaskParser.toJson(null))
Review Comment:
I see your point here. Will add a `TestScanTaskParser` to test the facade part, like task type field and unsupported task type.
You pointed out another issue in this test class. we should test both the old/deprecated public API (via `FileScanTaskParser.toJson/fromJson`) and new `ScanTaskParser` public API.
But this class still need to be changed as we are not going to add `FileScanTask` testing in the `TestScanTaskParser` facade test class.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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
Re: [PR] Core: add a new task-type field to task JSON serialization. add data task JSON serialization imp. [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #9728:
URL: https://github.com/apache/iceberg/pull/9728#discussion_r1526491086
##########
core/src/main/java/org/apache/iceberg/ScanTaskParser.java:
##########
@@ -0,0 +1,104 @@
+/*
+ * 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;
+
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.databind.JsonNode;
+import java.io.IOException;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.base.Strings;
+import org.apache.iceberg.util.JsonUtil;
+
+public class ScanTaskParser {
+ private static final String TASK_TYPE = "task-type";
+
+ private enum TaskType {
+ FILE_SCAN_TASK("file-scan-task"),
+ DATA_TASK("data-task");
+
+ private final String value;
+
+ TaskType(String value) {
+ this.value = value;
+ }
+
+ public static TaskType fromValue(String value) {
+ Preconditions.checkArgument(value != null, "Task type is null");
+ if (FILE_SCAN_TASK.value().equalsIgnoreCase(value)) {
+ return FILE_SCAN_TASK;
+ } else if (DATA_TASK.value().equalsIgnoreCase(value)) {
+ return DATA_TASK;
+ } else {
+ throw new IllegalArgumentException("Unknown task type: " + value);
+ }
+ }
+
+ public String value() {
+ return value;
+ }
+ }
+
+ private ScanTaskParser() {}
+
+ public static String toJson(FileScanTask fileScanTask) {
+ Preconditions.checkArgument(fileScanTask != null, "Invalid file scan task: null");
+ return JsonUtil.generate(generator -> toJson(fileScanTask, generator), false);
+ }
+
+ public static FileScanTask fromJson(String json, boolean caseSensitive) {
+ Preconditions.checkArgument(json != null, "Invalid JSON string for file scan task: null");
+ return JsonUtil.parse(json, node -> fromJson(node, caseSensitive));
+ }
+
+ private static void toJson(FileScanTask fileScanTask, JsonGenerator generator)
+ throws IOException {
+ generator.writeStartObject();
+
+ if (fileScanTask instanceof StaticDataTask) {
+ generator.writeStringField(TASK_TYPE, TaskType.DATA_TASK.value());
+ DataTaskParser.toJson((StaticDataTask) fileScanTask, generator);
+ } else if (fileScanTask instanceof BaseFileScanTask
+ || fileScanTask instanceof BaseFileScanTask.SplitScanTask) {
+ generator.writeStringField(TASK_TYPE, TaskType.FILE_SCAN_TASK.value());
+ FileScanTaskParser.toJson(fileScanTask, generator);
+ } else {
+ throw new UnsupportedOperationException(
+ "Unsupported task type: " + fileScanTask.getClass().getCanonicalName());
+ }
+
+ generator.writeEndObject();
+ }
+
+ private static FileScanTask fromJson(JsonNode jsonNode, boolean caseSensitive) {
+ TaskType taskType = TaskType.FILE_SCAN_TASK;
+ String taskTypeStr = JsonUtil.getStringOrNull(TASK_TYPE, jsonNode);
+ if (!Strings.isNullOrEmpty(taskTypeStr)) {
+ taskType = TaskType.fromValue(taskTypeStr);
+ }
+
+ switch (taskType) {
+ case FILE_SCAN_TASK:
+ return FileScanTaskParser.fromJson(jsonNode, caseSensitive);
+ case DATA_TASK:
+ return DataTaskParser.fromJson(jsonNode);
+ default:
+ throw new UnsupportedOperationException("Unsupported task type: " + taskType.value());
Review Comment:
same above
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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
Re: [PR] Core: add a new task-type field to task JSON serialization. add data task JSON serialization imp. [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #9728:
URL: https://github.com/apache/iceberg/pull/9728#discussion_r1583006389
##########
core/src/test/java/org/apache/iceberg/TestDataTaskParser.java:
##########
@@ -0,0 +1,249 @@
+/*
+ * 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;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.io.StringWriter;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.util.JsonUtil;
+import org.junit.jupiter.api.Test;
+
+public class TestDataTaskParser {
+ @Test
+ public void nullCheck() throws Exception {
+ StringWriter writer = new StringWriter();
+ JsonGenerator generator = JsonUtil.factory().createGenerator(writer);
+
+ assertThatThrownBy(() -> DataTaskParser.toJson(null, generator))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("Invalid data task: null");
+
+ assertThatThrownBy(() -> DataTaskParser.toJson((StaticDataTask) createDataTask(), null))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("Invalid JSON generator: null");
+
+ assertThatThrownBy(() -> DataTaskParser.fromJson(null))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("Invalid JSON node for data task: null");
+ }
+
+ @Test
+ public void invalidJsonNode() throws Exception {
+ String jsonStr = "{\"str\":\"1\", \"arr\":[]}";
+ ObjectMapper mapper = new ObjectMapper();
+ JsonNode rootNode = mapper.reader().readTree(jsonStr);
+
+ assertThatThrownBy(() -> DataTaskParser.fromJson(rootNode.get("str")))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Invalid JSON node for data task: non-object ");
+
+ assertThatThrownBy(() -> DataTaskParser.fromJson(rootNode.get("arr")))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Invalid JSON node for data task: non-object ");
+ }
+
+ @Test
+ public void missingFields() throws Exception {
+ ObjectMapper mapper = new ObjectMapper();
+
+ String missingSchemaStr = "{}";
+ JsonNode missingSchemaNode = mapper.reader().readTree(missingSchemaStr);
+ assertThatThrownBy(() -> DataTaskParser.fromJson(missingSchemaNode))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Cannot parse missing field: schema");
+
+ String missingProjectionStr =
+ "{"
+ + "\"schema\":{\"type\":\"struct\",\"schema-id\":0,"
+ + "\"fields\":[{\"id\":1,\"name\":\"committed_at\",\"required\":true,\"type\":\"timestamptz\"}]}"
+ + "}";
+ JsonNode missingProjectionNode = mapper.reader().readTree(missingProjectionStr);
+ assertThatThrownBy(() -> DataTaskParser.fromJson(missingProjectionNode))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Cannot parse missing field: projection");
+
+ String missingMetadataFileStr =
+ "{"
+ + "\"schema\":{\"type\":\"struct\",\"schema-id\":0,"
+ + "\"fields\":[{\"id\":1,\"name\":\"committed_at\",\"required\":true,\"type\":\"timestamptz\"}]},"
+ + "\"projection\":{\"type\":\"struct\",\"schema-id\":0,"
+ + "\"fields\":[{\"id\":1,\"name\":\"committed_at\",\"required\":true,\"type\":\"timestamptz\"}]}"
+ + "}";
+ JsonNode missingMetadataFileNode = mapper.reader().readTree(missingMetadataFileStr);
+ assertThatThrownBy(() -> DataTaskParser.fromJson(missingMetadataFileNode))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Cannot parse missing field: metadata-file");
+
+ String missingTableRowsStr =
+ "{\"task-type\":\"data-task\","
+ + "\"schema\":{\"type\":\"struct\",\"schema-id\":0,"
+ + "\"fields\":[{\"id\":1,\"name\":\"committed_at\",\"required\":true,\"type\":\"timestamptz\"},"
+ + "{\"id\":2,\"name\":\"snapshot_id\",\"required\":true,\"type\":\"long\"},"
+ + "{\"id\":3,\"name\":\"parent_id\",\"required\":false,\"type\":\"long\"},"
+ + "{\"id\":4,\"name\":\"operation\",\"required\":false,\"type\":\"string\"},"
+ + "{\"id\":5,\"name\":\"manifest_list\",\"required\":false,\"type\":\"string\"},"
+ + "{\"id\":6,\"name\":\"summary\",\"required\":false,\"type\":{\"type\":\"map\","
+ + "\"key-id\":7,\"key\":\"string\",\"value-id\":8,"
+ + "\"value\":\"string\",\"value-required\":true}}]},"
+ + "\"projection\":{\"type\":\"struct\",\"schema-id\":0,"
+ + "\"fields\":[{\"id\":1,\"name\":\"committed_at\",\"required\":true,\"type\":\"timestamptz\"},"
+ + "{\"id\":2,\"name\":\"snapshot_id\",\"required\":true,\"type\":\"long\"},"
+ + "{\"id\":3,\"name\":\"parent_id\",\"required\":false,\"type\":\"long\"},"
+ + "{\"id\":4,\"name\":\"operation\",\"required\":false,\"type\":\"string\"},"
+ + "{\"id\":5,\"name\":\"manifest_list\",\"required\":false,\"type\":\"string\"},"
+ + "{\"id\":6,\"name\":\"summary\",\"required\":false,\"type\":{\"type\":\"map\","
+ + "\"key-id\":7,\"key\":\"string\",\"value-id\":8,"
+ + "\"value\":\"string\",\"value-required\":true}}]},"
+ + "\"metadata-file\":{\"spec-id\":0,\"content\":\"DATA\","
+ + "\"file-path\":\"/tmp/metadata2.json\","
+ + "\"file-format\":\"METADATA\",\"partition\":{},"
+ + "\"file-size-in-bytes\":0,\"record-count\":2,\"sort-order-id\":0}"
+ + "}";
+ JsonNode missingTableRowsNode = mapper.reader().readTree(missingTableRowsStr);
+ assertThatThrownBy(() -> DataTaskParser.fromJson(missingTableRowsNode))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Cannot parse missing field: rows");
+ }
+
+ @Test
+ public void roundTripSerde() {
+ StaticDataTask dataTask = (StaticDataTask) createDataTask();
+ String jsonStr = ScanTaskParser.toJson(dataTask);
+ assertThat(jsonStr).isEqualTo(snapshotsDataTaskJson());
+ StaticDataTask deserializedTask = (StaticDataTask) ScanTaskParser.fromJson(jsonStr, true);
+ assertDataTaskEquals(dataTask, deserializedTask);
+ }
+
+ private DataTask createDataTask() {
+ Map<String, String> summary1 =
+ ImmutableMap.of(
+ "added-data-files", "1",
+ "added-records", "1",
+ "added-files-size", "10",
+ "changed-partition-count", "1",
+ "total-records", "1",
+ "total-files-size", "10",
+ "total-data-files", "1",
+ "total-delete-files", "0",
+ "total-position-deletes", "0",
+ "total-equality-deletes", "0");
+
+ Map<String, String> summary2 =
+ ImmutableMap.of(
+ "added-data-files", "1",
+ "added-records", "1",
+ "added-files-size", "10",
+ "changed-partition-count", "1",
+ "total-records", "2",
+ "total-files-size", "20",
+ "total-data-files", "2",
+ "total-delete-files", "0",
+ "total-position-deletes", "0",
+ "total-equality-deletes", "0");
+
+ List<Snapshot> snapshots =
+ Arrays.asList(
+ new BaseSnapshot(
+ 1L, 1L, null, 1234567890000L, "append", summary1, 1, "file:/tmp/manifest1.avro"),
+ new BaseSnapshot(
+ 2L, 2L, 1L, 9876543210000L, "append", summary2, 1, "file:/tmp/manifest2.avro"));
+
+ return StaticDataTask.of(
+ Files.localInput("file:/tmp/metadata2.json"),
+ SnapshotsTable.SNAPSHOT_SCHEMA,
+ SnapshotsTable.SNAPSHOT_SCHEMA,
+ snapshots,
+ SnapshotsTable::snapshotToRow);
+ }
+
+ private String snapshotsDataTaskJson() {
+ return "{\"task-type\":\"data-task\","
+ + "\"schema\":{\"type\":\"struct\",\"schema-id\":0,"
+ + "\"fields\":[{\"id\":1,\"name\":\"committed_at\",\"required\":true,\"type\":\"timestamptz\"},"
+ + "{\"id\":2,\"name\":\"snapshot_id\",\"required\":true,\"type\":\"long\"},"
+ + "{\"id\":3,\"name\":\"parent_id\",\"required\":false,\"type\":\"long\"},"
+ + "{\"id\":4,\"name\":\"operation\",\"required\":false,\"type\":\"string\"},"
+ + "{\"id\":5,\"name\":\"manifest_list\",\"required\":false,\"type\":\"string\"},"
+ + "{\"id\":6,\"name\":\"summary\",\"required\":false,\"type\":{\"type\":\"map\","
+ + "\"key-id\":7,\"key\":\"string\",\"value-id\":8,"
+ + "\"value\":\"string\",\"value-required\":true}}]},"
+ + "\"projection\":{\"type\":\"struct\",\"schema-id\":0,"
+ + "\"fields\":[{\"id\":1,\"name\":\"committed_at\",\"required\":true,\"type\":\"timestamptz\"},"
+ + "{\"id\":2,\"name\":\"snapshot_id\",\"required\":true,\"type\":\"long\"},"
+ + "{\"id\":3,\"name\":\"parent_id\",\"required\":false,\"type\":\"long\"},"
+ + "{\"id\":4,\"name\":\"operation\",\"required\":false,\"type\":\"string\"},"
+ + "{\"id\":5,\"name\":\"manifest_list\",\"required\":false,\"type\":\"string\"},"
+ + "{\"id\":6,\"name\":\"summary\",\"required\":false,\"type\":{\"type\":\"map\","
+ + "\"key-id\":7,\"key\":\"string\",\"value-id\":8,"
+ + "\"value\":\"string\",\"value-required\":true}}]},"
+ + "\"metadata-file\":{\"spec-id\":0,\"content\":\"DATA\","
+ + "\"file-path\":\"/tmp/metadata2.json\","
+ + "\"file-format\":\"METADATA\",\"partition\":{},"
+ + "\"file-size-in-bytes\":0,\"record-count\":2,\"sort-order-id\":0},"
+ + "\"rows\":[{\"1\":\"2009-02-13T23:31:30+00:00\",\"2\":1,\"4\":\"append\","
+ + "\"5\":\"file:/tmp/manifest1.avro\","
+ + "\"6\":{\"keys\":[\"added-data-files\",\"added-records\",\"added-files-size\",\"changed-partition-count\","
+ + "\"total-records\",\"total-files-size\",\"total-data-files\",\"total-delete-files\","
+ + "\"total-position-deletes\",\"total-equality-deletes\"],"
+ + "\"values\":[\"1\",\"1\",\"10\",\"1\",\"1\",\"10\",\"1\",\"0\",\"0\",\"0\"]}},"
+ + "{\"1\":\"2282-12-22T20:13:30+00:00\",\"2\":2,\"3\":1,\"4\":\"append\","
+ + "\"5\":\"file:/tmp/manifest2.avro\","
+ + "\"6\":{\"keys\":[\"added-data-files\",\"added-records\",\"added-files-size\",\"changed-partition-count\","
+ + "\"total-records\",\"total-files-size\",\"total-data-files\",\"total-delete-files\","
+ + "\"total-position-deletes\",\"total-equality-deletes\"],"
+ + "\"values\":[\"1\",\"1\",\"10\",\"1\",\"2\",\"20\",\"2\",\"0\",\"0\",\"0\"]}}]}";
+ }
+
+ private void assertDataTaskEquals(StaticDataTask expected, StaticDataTask actual) {
+ assertThat(expected.schema().asStruct())
+ .as("Schema should match")
+ .isEqualTo(actual.schema().asStruct());
+
+ assertThat(expected.projectedSchema().asStruct())
+ .as("Projected schema should match")
+ .isEqualTo(actual.projectedSchema().asStruct());
+
+ TestContentFileParser.assertContentFileEquals(
+ expected.metadataFile(), actual.metadataFile(), PartitionSpec.unpartitioned());
+
+ List<StructLike> expectedRows = Lists.newArrayList(expected.rows());
+ List<StructLike> actualRows = Lists.newArrayList(actual.rows());
+ assertThat(actualRows).hasSize(expectedRows.size());
Review Comment:
```suggestion
assertThat(actualRows).hasSameSizeAs(expectedRows);
```
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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
Re: [PR] Spec, Core: add a new task-type field to task JSON serialization. add data task JSON serialization spec and imp. [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #9728:
URL: https://github.com/apache/iceberg/pull/9728#discussion_r1495209384
##########
format/spec.md:
##########
@@ -1237,17 +1237,36 @@ Content file (data or delete) is serialized as a JSON object according to the fo
| **`equality-ids`** |`JSON list of int: Field ids used to determine row equality in equality delete files`|`[1]`|
| **`sort-order-id`** |`JSON int`|`1`|
-### File Scan Task Serialization
-
-File scan task is serialized as a JSON object according to the following table.
-
-| Metadata field |JSON representation|Example|
-|--------------------------|--- |--- |
-| **`schema`** |`JSON object`|`See above, read schemas instead`|
-| **`spec`** |`JSON object`|`See above, read partition specs instead`|
-| **`data-file`** |`JSON object`|`See above, read content file instead`|
-| **`delete-files`** |`JSON list of objects`|`See above, read content file instead`|
-| **`residual-filter`** |`JSON object: residual filter expression`|`{"type":"eq","term":"id","value":1}`|
+### Task Serialization
Review Comment:
for now, I have reverted the spec change. we still need to decide if/how we can remove the existing spec section on file scan task JSON serialization.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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
Re: [PR] Spec, Core: add a new task-type field to task JSON serialization. add data task JSON serialization spec and imp. [iceberg]
Posted by "emkornfield (via GitHub)" <gi...@apache.org>.
emkornfield commented on code in PR #9728:
URL: https://github.com/apache/iceberg/pull/9728#discussion_r1493059452
##########
format/spec.md:
##########
@@ -1237,17 +1237,36 @@ Content file (data or delete) is serialized as a JSON object according to the fo
| **`equality-ids`** |`JSON list of int: Field ids used to determine row equality in equality delete files`|`[1]`|
| **`sort-order-id`** |`JSON int`|`1`|
-### File Scan Task Serialization
-
-File scan task is serialized as a JSON object according to the following table.
-
-| Metadata field |JSON representation|Example|
-|--------------------------|--- |--- |
-| **`schema`** |`JSON object`|`See above, read schemas instead`|
-| **`spec`** |`JSON object`|`See above, read partition specs instead`|
-| **`data-file`** |`JSON object`|`See above, read content file instead`|
-| **`delete-files`** |`JSON list of objects`|`See above, read content file instead`|
-| **`residual-filter`** |`JSON object: residual filter expression`|`{"type":"eq","term":"id","value":1}`|
+### Task Serialization
Review Comment:
What state is being serialized? How does it relate to reading tables?
Does the flink use-case require standardization here or is it an implementation detail of flink?
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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
Re: [PR] Core: add a new task-type field to task JSON serialization. add data task JSON serialization implementation. [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #9728:
URL: https://github.com/apache/iceberg/pull/9728#discussion_r1645245721
##########
core/src/main/java/org/apache/iceberg/SnapshotsTable.java:
##########
@@ -27,7 +28,8 @@
* <p>This does not include snapshots that have been expired using {@link ExpireSnapshots}.
*/
public class SnapshotsTable extends BaseMetadataTable {
- private static final Schema SNAPSHOT_SCHEMA =
+ @VisibleForTesting
+ static final Schema SNAPSHOT_SCHEMA =
Review Comment:
copied the SnapshotsTable schema to the test class.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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
Re: [PR] Core: add a new task-type field to task JSON serialization. add data task JSON serialization imp. [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on PR #9728:
URL: https://github.com/apache/iceberg/pull/9728#issuecomment-2026251811
@nastra can you help take another look?
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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
Re: [PR] Core: add a new task-type field to task JSON serialization. add data task JSON serialization imp. [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #9728:
URL: https://github.com/apache/iceberg/pull/9728#discussion_r1576025475
##########
core/src/test/java/org/apache/iceberg/TestFileScanTaskParser.java:
##########
@@ -33,23 +33,64 @@ public void testNullArguments() {
.isInstanceOf(IllegalArgumentException.class)
.hasMessage("Invalid file scan task: null");
- Assertions.assertThatThrownBy(() -> FileScanTaskParser.fromJson(null, true))
+ Assertions.assertThatThrownBy(() -> FileScanTaskParser.fromJson((String) null, true))
.isInstanceOf(IllegalArgumentException.class)
.hasMessage("Invalid JSON string for file scan task: null");
+
+ Assertions.assertThatThrownBy(() -> ScanTaskParser.toJson(null))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("Invalid scan task: null");
+
+ Assertions.assertThatThrownBy(() -> ScanTaskParser.fromJson(null, true))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("Invalid JSON string for scan task: null");
}
@ParameterizedTest
@ValueSource(booleans = {true, false})
- public void testParser(boolean caseSensitive) {
+ public void testFileScanTaskParser(boolean caseSensitive) {
PartitionSpec spec = TestBase.SPEC;
- FileScanTask fileScanTask = createScanTask(spec, caseSensitive);
+ FileScanTask fileScanTask = createFileScanTask(spec, caseSensitive);
String jsonStr = FileScanTaskParser.toJson(fileScanTask);
- Assertions.assertThat(jsonStr).isEqualTo(expectedFileScanTaskJson());
+ Assertions.assertThat(jsonStr).isEqualTo(fileScanTaskJsonWithoutTaskType());
FileScanTask deserializedTask = FileScanTaskParser.fromJson(jsonStr, caseSensitive);
assertFileScanTaskEquals(fileScanTask, deserializedTask, spec, caseSensitive);
}
- private FileScanTask createScanTask(PartitionSpec spec, boolean caseSensitive) {
+ /** Test backward compatibility where task-type field is absent from the JSON string */
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testFileScanTaskParserWithoutTaskTypeField(boolean caseSensitive) {
+ PartitionSpec spec = TestBase.SPEC;
+ FileScanTask fileScanTask = createFileScanTask(spec, caseSensitive);
+ FileScanTask deserializedTask =
+ FileScanTaskParser.fromJson(fileScanTaskJsonWithoutTaskType(), caseSensitive);
+ assertFileScanTaskEquals(fileScanTask, deserializedTask, spec, caseSensitive);
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testScanTaskParser(boolean caseSensitive) {
Review Comment:
imho this test should live in `TestScanTaskParser` as it's focusing on the `ScanTaskParser`. Same for `testScanTaskParserWithoutTaskTypeField`
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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
Re: [PR] Core: add a new task-type field to task JSON serialization. add data task JSON serialization imp. [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #9728:
URL: https://github.com/apache/iceberg/pull/9728#discussion_r1583003922
##########
core/src/test/java/org/apache/iceberg/TestFileScanTaskParser.java:
##########
@@ -84,20 +127,38 @@ private String expectedFileScanTaskJson() {
+ "\"residual-filter\":{\"type\":\"eq\",\"term\":\"id\",\"value\":1}}";
}
+ private String fileScanTaskJson() {
+ return "{\"task-type\":\"file-scan-task\","
+ + "\"schema\":{\"type\":\"struct\",\"schema-id\":0,\"fields\":["
+ + "{\"id\":3,\"name\":\"id\",\"required\":true,\"type\":\"int\"},"
+ + "{\"id\":4,\"name\":\"data\",\"required\":true,\"type\":\"string\"}]},"
+ + "\"spec\":{\"spec-id\":0,\"fields\":[{\"name\":\"data_bucket\","
+ + "\"transform\":\"bucket[16]\",\"source-id\":4,\"field-id\":1000}]},"
+ + "\"data-file\":{\"spec-id\":0,\"content\":\"DATA\",\"file-path\":\"/path/to/data-a.parquet\","
+ + "\"file-format\":\"PARQUET\",\"partition\":{\"1000\":0},"
+ + "\"file-size-in-bytes\":10,\"record-count\":1,\"sort-order-id\":0},"
+ + "\"start\":0,\"length\":10,"
+ + "\"delete-files\":[{\"spec-id\":0,\"content\":\"POSITION_DELETES\","
+ + "\"file-path\":\"/path/to/data-a-deletes.parquet\",\"file-format\":\"PARQUET\","
+ + "\"partition\":{\"1000\":0},\"file-size-in-bytes\":10,\"record-count\":1},"
+ + "{\"spec-id\":0,\"content\":\"EQUALITY_DELETES\",\"file-path\":\"/path/to/data-a2-deletes.parquet\","
+ + "\"file-format\":\"PARQUET\",\"partition\":{\"1000\":0},\"file-size-in-bytes\":10,"
+ + "\"record-count\":1,\"equality-ids\":[1],\"sort-order-id\":0}],"
+ + "\"residual-filter\":{\"type\":\"eq\",\"term\":\"id\",\"value\":1}}";
+ }
+
private static void assertFileScanTaskEquals(
FileScanTask expected, FileScanTask actual, PartitionSpec spec, boolean caseSensitive) {
TestContentFileParser.assertContentFileEquals(expected.file(), actual.file(), spec);
- Assertions.assertThat(actual.deletes()).hasSameSizeAs(expected.deletes());
+ assertThat(actual.deletes()).hasSameSizeAs(expected.deletes());
for (int pos = 0; pos < expected.deletes().size(); ++pos) {
TestContentFileParser.assertContentFileEquals(
expected.deletes().get(pos), actual.deletes().get(pos), spec);
}
- Assertions.assertThat(expected.schema().sameSchema(actual.schema()))
- .as("Schema should match")
- .isTrue();
- Assertions.assertThat(actual.spec()).isEqualTo(expected.spec());
- Assertions.assertThat(
+ assertThat(expected.schema().sameSchema(actual.schema())).as("Schema should match").isTrue();
Review Comment:
```suggestion
assertThat(actual.schema().asStruct()).isEqualTo(expected.schema().asStruct());
```
if the assertion fails, then this will show where the schema mismatch is
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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
Re: [PR] Core: add a new task-type field to task JSON serialization. add data task JSON serialization imp. [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on PR #9728:
URL: https://github.com/apache/iceberg/pull/9728#issuecomment-1995046941
@nastra @pvary can you help review? spec change/revert has been moved to a separate PR: https://github.com/apache/iceberg/pull/9771/files.
This is blocking Flink moving to the FLIP-27 source as default. will need another PR to complete the metadata queries support.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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
Re: [PR] Core: add a new task-type field to task JSON serialization. add data task JSON serialization imp. [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #9728:
URL: https://github.com/apache/iceberg/pull/9728#discussion_r1524825126
##########
core/src/main/java/org/apache/iceberg/ScanTaskParser.java:
##########
@@ -0,0 +1,104 @@
+/*
+ * 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;
+
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.databind.JsonNode;
+import java.io.IOException;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.base.Strings;
+import org.apache.iceberg.util.JsonUtil;
+
+public class ScanTaskParser {
+ private static final String TASK_TYPE = "task-type";
+
+ private enum TaskType {
+ FILE_SCAN_TASK("file-scan-task"),
+ DATA_TASK("data-task");
+
+ private final String value;
+
+ TaskType(String value) {
+ this.value = value;
+ }
+
+ public static TaskType fromValue(String value) {
+ Preconditions.checkArgument(value != null, "Task type is null");
Review Comment:
```suggestion
Preconditions.checkArgument(value != null, "Invalid task type: null");
```
##########
core/src/main/java/org/apache/iceberg/ScanTaskParser.java:
##########
@@ -0,0 +1,104 @@
+/*
+ * 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;
+
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.databind.JsonNode;
+import java.io.IOException;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.base.Strings;
+import org.apache.iceberg.util.JsonUtil;
+
+public class ScanTaskParser {
+ private static final String TASK_TYPE = "task-type";
+
+ private enum TaskType {
+ FILE_SCAN_TASK("file-scan-task"),
Review Comment:
I think this should have an `UNKNOWN` for forward/backward compatibility. Imagine a client/server that use different Iceberg versions and new task types are being added over time.
##########
core/src/main/java/org/apache/iceberg/DataTaskParser.java:
##########
@@ -0,0 +1,80 @@
+/*
+ * 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;
+
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.databind.JsonNode;
+import java.io.IOException;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.util.JsonUtil;
+
+class DataTaskParser {
+ private static final String SCHEMA = "schema";
+ private static final String PROJECTED_SCHEMA = "projection";
+ private static final String METADATA_FILE = "metadata-file";
+ private static final String ROWS = "rows";
+
+ private DataTaskParser() {}
+
+ static void toJson(StaticDataTask dataTask, JsonGenerator generator) throws IOException {
+ Preconditions.checkArgument(dataTask != null, "Invalid data task: null");
+ Preconditions.checkArgument(generator != null, "Invalid JSON generator: null");
+
+ generator.writeFieldName(SCHEMA);
+ SchemaParser.toJson(dataTask.schema(), generator);
+
+ generator.writeFieldName(PROJECTED_SCHEMA);
+ SchemaParser.toJson(dataTask.projectedSchema(), generator);
+
+ generator.writeFieldName(METADATA_FILE);
+ ContentFileParser.toJson(dataTask.metadataFile(), PartitionSpec.unpartitioned(), generator);
+
+ generator.writeArrayFieldStart(ROWS);
+ for (StructLike row : dataTask.tableRows()) {
+ SingleValueParser.toJson(dataTask.schema().asStruct(), row, generator);
+ }
+
+ generator.writeEndArray();
+ }
+
+ static StaticDataTask fromJson(JsonNode jsonNode) {
+ Preconditions.checkArgument(jsonNode != null, "Invalid JSON node for data task: null");
+ Preconditions.checkArgument(
+ jsonNode.isObject(), "Invalid JSON node for data task: non-object (%s)", jsonNode);
+
+ Schema schema = SchemaParser.fromJson(JsonUtil.get(SCHEMA, jsonNode));
+ Schema projectedSchema = SchemaParser.fromJson(JsonUtil.get(PROJECTED_SCHEMA, jsonNode));
+ DataFile metadataFile =
+ (DataFile)
+ ContentFileParser.fromJson(
+ JsonUtil.get(METADATA_FILE, jsonNode), PartitionSpec.unpartitioned());
+
+ JsonNode rowsArray = jsonNode.get(ROWS);
Review Comment:
please use `JsonUtil.get` in such places, as it is a required field and `JsonUtil.get` will throw a descriptive error msg if this field is missing
##########
core/src/main/java/org/apache/iceberg/ScanTaskParser.java:
##########
@@ -0,0 +1,104 @@
+/*
+ * 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;
+
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.databind.JsonNode;
+import java.io.IOException;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.base.Strings;
+import org.apache.iceberg.util.JsonUtil;
+
+public class ScanTaskParser {
+ private static final String TASK_TYPE = "task-type";
+
+ private enum TaskType {
+ FILE_SCAN_TASK("file-scan-task"),
+ DATA_TASK("data-task");
+
+ private final String value;
+
+ TaskType(String value) {
+ this.value = value;
+ }
+
+ public static TaskType fromValue(String value) {
+ Preconditions.checkArgument(value != null, "Task type is null");
+ if (FILE_SCAN_TASK.value().equalsIgnoreCase(value)) {
+ return FILE_SCAN_TASK;
+ } else if (DATA_TASK.value().equalsIgnoreCase(value)) {
+ return DATA_TASK;
+ } else {
+ throw new IllegalArgumentException("Unknown task type: " + value);
Review Comment:
this probably shouldn't fail but rather return `UNKNOWN`. See also #7145 where a similar issue has been addressed and I think we need to do the same thing here
##########
core/src/test/java/org/apache/iceberg/TestFileScanTaskParser.java:
##########
@@ -27,29 +27,41 @@
import org.junit.jupiter.params.provider.ValueSource;
public class TestFileScanTaskParser {
+
@Test
public void testNullArguments() {
- Assertions.assertThatThrownBy(() -> FileScanTaskParser.toJson(null))
+ Assertions.assertThatThrownBy(() -> ScanTaskParser.toJson(null))
Review Comment:
I don't think this test class should actually change at all. Since you added a new `ScanTaskParser`, it's best to also add a `TestScanTaskParser`. This test should stay the same, since someone could still use the Parser and the existing behavior of the parser shouldn't change IMO
##########
core/src/main/java/org/apache/iceberg/SnapshotsTable.java:
##########
@@ -27,7 +28,8 @@
* <p>This does not include snapshots that have been expired using {@link ExpireSnapshots}.
*/
public class SnapshotsTable extends BaseMetadataTable {
- private static final Schema SNAPSHOT_SCHEMA =
+ @VisibleForTesting
+ static final Schema SNAPSHOT_SCHEMA =
Review Comment:
could we avoid making this visible?
##########
core/src/main/java/org/apache/iceberg/ScanTaskParser.java:
##########
@@ -0,0 +1,104 @@
+/*
+ * 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;
+
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.databind.JsonNode;
+import java.io.IOException;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.base.Strings;
+import org.apache.iceberg.util.JsonUtil;
+
+public class ScanTaskParser {
+ private static final String TASK_TYPE = "task-type";
+
+ private enum TaskType {
+ FILE_SCAN_TASK("file-scan-task"),
+ DATA_TASK("data-task");
+
+ private final String value;
+
+ TaskType(String value) {
+ this.value = value;
+ }
+
+ public static TaskType fromValue(String value) {
+ Preconditions.checkArgument(value != null, "Task type is null");
+ if (FILE_SCAN_TASK.value().equalsIgnoreCase(value)) {
+ return FILE_SCAN_TASK;
+ } else if (DATA_TASK.value().equalsIgnoreCase(value)) {
+ return DATA_TASK;
+ } else {
+ throw new IllegalArgumentException("Unknown task type: " + value);
+ }
+ }
+
+ public String value() {
+ return value;
+ }
+ }
+
+ private ScanTaskParser() {}
+
+ public static String toJson(FileScanTask fileScanTask) {
+ Preconditions.checkArgument(fileScanTask != null, "Invalid file scan task: null");
+ return JsonUtil.generate(generator -> toJson(fileScanTask, generator), false);
+ }
+
+ public static FileScanTask fromJson(String json, boolean caseSensitive) {
+ Preconditions.checkArgument(json != null, "Invalid JSON string for file scan task: null");
+ return JsonUtil.parse(json, node -> fromJson(node, caseSensitive));
+ }
+
+ private static void toJson(FileScanTask fileScanTask, JsonGenerator generator)
+ throws IOException {
+ generator.writeStartObject();
+
+ if (fileScanTask instanceof StaticDataTask) {
+ generator.writeStringField(TASK_TYPE, TaskType.DATA_TASK.value());
+ DataTaskParser.toJson((StaticDataTask) fileScanTask, generator);
+ } else if (fileScanTask instanceof BaseFileScanTask
+ || fileScanTask instanceof BaseFileScanTask.SplitScanTask) {
+ generator.writeStringField(TASK_TYPE, TaskType.FILE_SCAN_TASK.value());
+ FileScanTaskParser.toJson(fileScanTask, generator);
+ } else {
+ throw new UnsupportedOperationException(
Review Comment:
I don't think this should fail for the same reason I mentioned further above. If you e.g. take a look at the `ReportMetricsRequestParser`, then it doesn't fail if if sees an unknown type and we would need to do the same handling here
##########
core/src/main/java/org/apache/iceberg/ScanTaskParser.java:
##########
@@ -0,0 +1,104 @@
+/*
+ * 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;
+
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.databind.JsonNode;
+import java.io.IOException;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.base.Strings;
+import org.apache.iceberg.util.JsonUtil;
+
+public class ScanTaskParser {
+ private static final String TASK_TYPE = "task-type";
+
+ private enum TaskType {
+ FILE_SCAN_TASK("file-scan-task"),
+ DATA_TASK("data-task");
+
+ private final String value;
+
+ TaskType(String value) {
+ this.value = value;
+ }
+
+ public static TaskType fromValue(String value) {
+ Preconditions.checkArgument(value != null, "Task type is null");
+ if (FILE_SCAN_TASK.value().equalsIgnoreCase(value)) {
+ return FILE_SCAN_TASK;
+ } else if (DATA_TASK.value().equalsIgnoreCase(value)) {
+ return DATA_TASK;
+ } else {
+ throw new IllegalArgumentException("Unknown task type: " + value);
+ }
+ }
+
+ public String value() {
Review Comment:
nit: maybe `typeName()`?
##########
core/src/test/java/org/apache/iceberg/TestDataTaskParser.java:
##########
@@ -0,0 +1,147 @@
+/*
+ * 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;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.assertj.core.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class TestDataTaskParser {
Review Comment:
this parser currently only tests the happy path. Please also add tests that
* check for null/empty json when calling to/from json methods
* check for missing fields
* check for invalid task type(s)
* check a full roundtrip (which is what you have currently here)
You can take a look at the testing patterns in `TestReportMetricsRequestParser` / `TestLoadViewResponseParser`
##########
core/src/main/java/org/apache/iceberg/DataTaskParser.java:
##########
@@ -0,0 +1,80 @@
+/*
+ * 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;
+
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.databind.JsonNode;
+import java.io.IOException;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.util.JsonUtil;
+
+class DataTaskParser {
+ private static final String SCHEMA = "schema";
+ private static final String PROJECTED_SCHEMA = "projection";
+ private static final String METADATA_FILE = "metadata-file";
+ private static final String ROWS = "rows";
+
+ private DataTaskParser() {}
+
+ static void toJson(StaticDataTask dataTask, JsonGenerator generator) throws IOException {
+ Preconditions.checkArgument(dataTask != null, "Invalid data task: null");
+ Preconditions.checkArgument(generator != null, "Invalid JSON generator: null");
+
+ generator.writeFieldName(SCHEMA);
+ SchemaParser.toJson(dataTask.schema(), generator);
+
+ generator.writeFieldName(PROJECTED_SCHEMA);
+ SchemaParser.toJson(dataTask.projectedSchema(), generator);
+
+ generator.writeFieldName(METADATA_FILE);
+ ContentFileParser.toJson(dataTask.metadataFile(), PartitionSpec.unpartitioned(), generator);
+
+ generator.writeArrayFieldStart(ROWS);
+ for (StructLike row : dataTask.tableRows()) {
Review Comment:
can `tableRows()` be empty?
##########
core/src/main/java/org/apache/iceberg/FileScanTaskParser.java:
##########
@@ -40,16 +40,32 @@ public class FileScanTaskParser {
private FileScanTaskParser() {}
+ /**
+ * Serialize file scan task to JSON string
+ *
+ * @deprecated Use {@link ScanTaskParser#toJson(FileScanTask)} instead; will be removed in 1.7.0
+ */
+ @Deprecated
public static String toJson(FileScanTask fileScanTask) {
- return JsonUtil.generate(
- generator -> FileScanTaskParser.toJson(fileScanTask, generator), false);
+ Preconditions.checkArgument(fileScanTask != null, "Invalid file scan task: null");
+ return JsonUtil.generate(generator -> toJson(fileScanTask, generator), false);
+ }
+
+ /**
+ * Deserialize file scan task from JSON string
+ *
+ * @deprecated Use {@link ScanTaskParser#fromJson(String, boolean)} instead; will be removed in
+ * 1.7.0
+ */
+ @Deprecated
+ public static FileScanTask fromJson(String json, boolean caseSensitive) {
+ Preconditions.checkArgument(json != null, "Invalid JSON string for file scan task: null");
+ return JsonUtil.parse(json, node -> fromJson(node, caseSensitive));
}
- private static void toJson(FileScanTask fileScanTask, JsonGenerator generator)
- throws IOException {
+ static void toJson(FileScanTask fileScanTask, JsonGenerator generator) throws IOException {
Preconditions.checkArgument(fileScanTask != null, "Invalid file scan task: null");
Preconditions.checkArgument(generator != null, "Invalid JSON generator: null");
- generator.writeStartObject();
Review Comment:
isn't this changing behavior when the parser is being used elsewhere?
##########
core/src/main/java/org/apache/iceberg/ScanTaskParser.java:
##########
@@ -0,0 +1,104 @@
+/*
+ * 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;
+
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.databind.JsonNode;
+import java.io.IOException;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.base.Strings;
+import org.apache.iceberg.util.JsonUtil;
+
+public class ScanTaskParser {
+ private static final String TASK_TYPE = "task-type";
+
+ private enum TaskType {
+ FILE_SCAN_TASK("file-scan-task"),
+ DATA_TASK("data-task");
+
+ private final String value;
+
+ TaskType(String value) {
+ this.value = value;
+ }
+
+ public static TaskType fromValue(String value) {
+ Preconditions.checkArgument(value != null, "Task type is null");
+ if (FILE_SCAN_TASK.value().equalsIgnoreCase(value)) {
+ return FILE_SCAN_TASK;
+ } else if (DATA_TASK.value().equalsIgnoreCase(value)) {
+ return DATA_TASK;
+ } else {
+ throw new IllegalArgumentException("Unknown task type: " + value);
+ }
+ }
+
+ public String value() {
+ return value;
+ }
+ }
+
+ private ScanTaskParser() {}
+
+ public static String toJson(FileScanTask fileScanTask) {
+ Preconditions.checkArgument(fileScanTask != null, "Invalid file scan task: null");
+ return JsonUtil.generate(generator -> toJson(fileScanTask, generator), false);
+ }
+
+ public static FileScanTask fromJson(String json, boolean caseSensitive) {
+ Preconditions.checkArgument(json != null, "Invalid JSON string for file scan task: null");
+ return JsonUtil.parse(json, node -> fromJson(node, caseSensitive));
+ }
+
+ private static void toJson(FileScanTask fileScanTask, JsonGenerator generator)
+ throws IOException {
+ generator.writeStartObject();
+
+ if (fileScanTask instanceof StaticDataTask) {
+ generator.writeStringField(TASK_TYPE, TaskType.DATA_TASK.value());
+ DataTaskParser.toJson((StaticDataTask) fileScanTask, generator);
+ } else if (fileScanTask instanceof BaseFileScanTask
+ || fileScanTask instanceof BaseFileScanTask.SplitScanTask) {
+ generator.writeStringField(TASK_TYPE, TaskType.FILE_SCAN_TASK.value());
+ FileScanTaskParser.toJson(fileScanTask, generator);
+ } else {
+ throw new UnsupportedOperationException(
+ "Unsupported task type: " + fileScanTask.getClass().getCanonicalName());
+ }
+
+ generator.writeEndObject();
+ }
+
+ private static FileScanTask fromJson(JsonNode jsonNode, boolean caseSensitive) {
+ TaskType taskType = TaskType.FILE_SCAN_TASK;
+ String taskTypeStr = JsonUtil.getStringOrNull(TASK_TYPE, jsonNode);
+ if (!Strings.isNullOrEmpty(taskTypeStr)) {
+ taskType = TaskType.fromValue(taskTypeStr);
+ }
+
+ switch (taskType) {
+ case FILE_SCAN_TASK:
+ return FileScanTaskParser.fromJson(jsonNode, caseSensitive);
+ case DATA_TASK:
+ return DataTaskParser.fromJson(jsonNode);
+ default:
+ throw new UnsupportedOperationException("Unsupported task type: " + taskType.value());
Review Comment:
this shouldn't fail and should be handled similarly to how it's done in `ReportMetricsRequestParser`
##########
core/src/main/java/org/apache/iceberg/DataTaskParser.java:
##########
@@ -0,0 +1,80 @@
+/*
+ * 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;
+
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.databind.JsonNode;
+import java.io.IOException;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.util.JsonUtil;
+
+class DataTaskParser {
+ private static final String SCHEMA = "schema";
+ private static final String PROJECTED_SCHEMA = "projection";
+ private static final String METADATA_FILE = "metadata-file";
+ private static final String ROWS = "rows";
+
+ private DataTaskParser() {}
+
+ static void toJson(StaticDataTask dataTask, JsonGenerator generator) throws IOException {
+ Preconditions.checkArgument(dataTask != null, "Invalid data task: null");
+ Preconditions.checkArgument(generator != null, "Invalid JSON generator: null");
Review Comment:
`TestDataTaskParser` needs tests to cover these null checks
##########
core/src/main/java/org/apache/iceberg/ScanTaskParser.java:
##########
@@ -0,0 +1,104 @@
+/*
+ * 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;
+
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.databind.JsonNode;
+import java.io.IOException;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.base.Strings;
+import org.apache.iceberg.util.JsonUtil;
+
+public class ScanTaskParser {
+ private static final String TASK_TYPE = "task-type";
+
+ private enum TaskType {
+ FILE_SCAN_TASK("file-scan-task"),
+ DATA_TASK("data-task");
+
+ private final String value;
+
+ TaskType(String value) {
+ this.value = value;
+ }
+
+ public static TaskType fromValue(String value) {
+ Preconditions.checkArgument(value != null, "Task type is null");
+ if (FILE_SCAN_TASK.value().equalsIgnoreCase(value)) {
+ return FILE_SCAN_TASK;
+ } else if (DATA_TASK.value().equalsIgnoreCase(value)) {
+ return DATA_TASK;
+ } else {
+ throw new IllegalArgumentException("Unknown task type: " + value);
+ }
+ }
+
+ public String value() {
+ return value;
+ }
+ }
+
+ private ScanTaskParser() {}
+
+ public static String toJson(FileScanTask fileScanTask) {
+ Preconditions.checkArgument(fileScanTask != null, "Invalid file scan task: null");
+ return JsonUtil.generate(generator -> toJson(fileScanTask, generator), false);
+ }
+
+ public static FileScanTask fromJson(String json, boolean caseSensitive) {
+ Preconditions.checkArgument(json != null, "Invalid JSON string for file scan task: null");
+ return JsonUtil.parse(json, node -> fromJson(node, caseSensitive));
+ }
+
+ private static void toJson(FileScanTask fileScanTask, JsonGenerator generator)
+ throws IOException {
+ generator.writeStartObject();
+
+ if (fileScanTask instanceof StaticDataTask) {
+ generator.writeStringField(TASK_TYPE, TaskType.DATA_TASK.value());
+ DataTaskParser.toJson((StaticDataTask) fileScanTask, generator);
+ } else if (fileScanTask instanceof BaseFileScanTask
+ || fileScanTask instanceof BaseFileScanTask.SplitScanTask) {
+ generator.writeStringField(TASK_TYPE, TaskType.FILE_SCAN_TASK.value());
+ FileScanTaskParser.toJson(fileScanTask, generator);
+ } else {
+ throw new UnsupportedOperationException(
+ "Unsupported task type: " + fileScanTask.getClass().getCanonicalName());
+ }
+
+ generator.writeEndObject();
+ }
+
+ private static FileScanTask fromJson(JsonNode jsonNode, boolean caseSensitive) {
+ TaskType taskType = TaskType.FILE_SCAN_TASK;
+ String taskTypeStr = JsonUtil.getStringOrNull(TASK_TYPE, jsonNode);
+ if (!Strings.isNullOrEmpty(taskTypeStr)) {
Review Comment:
```suggestion
if (null != taskTypeStr) {
```
I don't think this should support having an empty string in the `TASK_TYPE`. If the `TASK_TYPE` field is sent, then it must always be a non-empty string
##########
core/src/main/java/org/apache/iceberg/DataTaskParser.java:
##########
@@ -0,0 +1,80 @@
+/*
+ * 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;
+
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.databind.JsonNode;
+import java.io.IOException;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.util.JsonUtil;
+
+class DataTaskParser {
+ private static final String SCHEMA = "schema";
+ private static final String PROJECTED_SCHEMA = "projection";
+ private static final String METADATA_FILE = "metadata-file";
+ private static final String ROWS = "rows";
+
+ private DataTaskParser() {}
+
+ static void toJson(StaticDataTask dataTask, JsonGenerator generator) throws IOException {
+ Preconditions.checkArgument(dataTask != null, "Invalid data task: null");
+ Preconditions.checkArgument(generator != null, "Invalid JSON generator: null");
+
+ generator.writeFieldName(SCHEMA);
+ SchemaParser.toJson(dataTask.schema(), generator);
+
+ generator.writeFieldName(PROJECTED_SCHEMA);
+ SchemaParser.toJson(dataTask.projectedSchema(), generator);
+
+ generator.writeFieldName(METADATA_FILE);
+ ContentFileParser.toJson(dataTask.metadataFile(), PartitionSpec.unpartitioned(), generator);
+
+ generator.writeArrayFieldStart(ROWS);
+ for (StructLike row : dataTask.tableRows()) {
+ SingleValueParser.toJson(dataTask.schema().asStruct(), row, generator);
+ }
+
+ generator.writeEndArray();
+ }
+
+ static StaticDataTask fromJson(JsonNode jsonNode) {
+ Preconditions.checkArgument(jsonNode != null, "Invalid JSON node for data task: null");
+ Preconditions.checkArgument(
+ jsonNode.isObject(), "Invalid JSON node for data task: non-object (%s)", jsonNode);
Review Comment:
`TestDataTaskParser` needs tests to cover these null/object checks
##########
core/src/test/java/org/apache/iceberg/TestDataTaskParser.java:
##########
@@ -0,0 +1,147 @@
+/*
+ * 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;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.assertj.core.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class TestDataTaskParser {
+ @Test
+ public void testStaticDataTask() {
+ StaticDataTask dataTask = (StaticDataTask) createDataTask();
+ String jsonStr = ScanTaskParser.toJson(dataTask);
+ Assertions.assertThat(jsonStr).isEqualTo(snapshotsDataTaskJson());
+ StaticDataTask deserializedTask = (StaticDataTask) ScanTaskParser.fromJson(jsonStr, true);
+ assertDataTaskEquals(dataTask, deserializedTask);
+ }
+
+ private DataTask createDataTask() {
+ Map<String, String> summary1 =
+ ImmutableMap.of(
+ "added-data-files", "1",
+ "added-records", "1",
+ "added-files-size", "10",
+ "changed-partition-count", "1",
+ "total-records", "1",
+ "total-files-size", "10",
+ "total-data-files", "1",
+ "total-delete-files", "0",
+ "total-position-deletes", "0",
+ "total-equality-deletes", "0");
+
+ Map<String, String> summary2 =
+ ImmutableMap.of(
+ "added-data-files", "1",
+ "added-records", "1",
+ "added-files-size", "10",
+ "changed-partition-count", "1",
+ "total-records", "2",
+ "total-files-size", "20",
+ "total-data-files", "2",
+ "total-delete-files", "0",
+ "total-position-deletes", "0",
+ "total-equality-deletes", "0");
+
+ List<Snapshot> snapshots =
+ Arrays.asList(
+ new BaseSnapshot(
+ 1L, 1L, null, 1234567890000L, "append", summary1, 1, "file:/tmp/manifest1.avro"),
+ new BaseSnapshot(
+ 2L, 2L, 1L, 9876543210000L, "append", summary2, 1, "file:/tmp/manifest2.avro"));
+
+ return StaticDataTask.of(
+ Files.localInput("file:/tmp/metadata2.json"),
+ SnapshotsTable.SNAPSHOT_SCHEMA,
+ SnapshotsTable.SNAPSHOT_SCHEMA,
+ snapshots,
+ SnapshotsTable::snapshotToRow);
+ }
+
+ private String snapshotsDataTaskJson() {
+ return "{\"task-type\":\"data-task\","
+ + "\"schema\":{\"type\":\"struct\",\"schema-id\":0,"
+ + "\"fields\":[{\"id\":1,\"name\":\"committed_at\",\"required\":true,\"type\":\"timestamptz\"},"
+ + "{\"id\":2,\"name\":\"snapshot_id\",\"required\":true,\"type\":\"long\"},"
+ + "{\"id\":3,\"name\":\"parent_id\",\"required\":false,\"type\":\"long\"},"
+ + "{\"id\":4,\"name\":\"operation\",\"required\":false,\"type\":\"string\"},"
+ + "{\"id\":5,\"name\":\"manifest_list\",\"required\":false,\"type\":\"string\"},"
+ + "{\"id\":6,\"name\":\"summary\",\"required\":false,\"type\":{\"type\":\"map\","
+ + "\"key-id\":7,\"key\":\"string\",\"value-id\":8,"
+ + "\"value\":\"string\",\"value-required\":true}}]},"
+ + "\"projection\":{\"type\":\"struct\",\"schema-id\":0,"
+ + "\"fields\":[{\"id\":1,\"name\":\"committed_at\",\"required\":true,\"type\":\"timestamptz\"},"
+ + "{\"id\":2,\"name\":\"snapshot_id\",\"required\":true,\"type\":\"long\"},"
+ + "{\"id\":3,\"name\":\"parent_id\",\"required\":false,\"type\":\"long\"},"
+ + "{\"id\":4,\"name\":\"operation\",\"required\":false,\"type\":\"string\"},"
+ + "{\"id\":5,\"name\":\"manifest_list\",\"required\":false,\"type\":\"string\"},"
+ + "{\"id\":6,\"name\":\"summary\",\"required\":false,\"type\":{\"type\":\"map\","
+ + "\"key-id\":7,\"key\":\"string\",\"value-id\":8,"
+ + "\"value\":\"string\",\"value-required\":true}}]},"
+ + "\"metadata-file\":{\"spec-id\":0,\"content\":\"DATA\","
+ + "\"file-path\":\"/tmp/metadata2.json\","
+ + "\"file-format\":\"METADATA\",\"partition\":{},"
+ + "\"file-size-in-bytes\":0,\"record-count\":2,\"sort-order-id\":0},"
+ + "\"rows\":[{\"1\":\"2009-02-13T23:31:30+00:00\",\"2\":1,\"4\":\"append\","
+ + "\"5\":\"file:/tmp/manifest1.avro\","
+ + "\"6\":{\"keys\":[\"added-data-files\",\"added-records\",\"added-files-size\",\"changed-partition-count\","
+ + "\"total-records\",\"total-files-size\",\"total-data-files\",\"total-delete-files\","
+ + "\"total-position-deletes\",\"total-equality-deletes\"],"
+ + "\"values\":[\"1\",\"1\",\"10\",\"1\",\"1\",\"10\",\"1\",\"0\",\"0\",\"0\"]}},"
+ + "{\"1\":\"2282-12-22T20:13:30+00:00\",\"2\":2,\"3\":1,\"4\":\"append\","
+ + "\"5\":\"file:/tmp/manifest2.avro\","
+ + "\"6\":{\"keys\":[\"added-data-files\",\"added-records\",\"added-files-size\",\"changed-partition-count\","
+ + "\"total-records\",\"total-files-size\",\"total-data-files\",\"total-delete-files\","
+ + "\"total-position-deletes\",\"total-equality-deletes\"],"
+ + "\"values\":[\"1\",\"1\",\"10\",\"1\",\"2\",\"20\",\"2\",\"0\",\"0\",\"0\"]}}]}";
+ }
+
+ private void assertDataTaskEquals(StaticDataTask expected, StaticDataTask actual) {
+ Assertions.assertThat(expected.schema().sameSchema(actual.schema()))
+ .as("Schema should match")
+ .isTrue();
+
+ Assertions.assertThat(expected.projectedSchema().sameSchema(actual.projectedSchema()))
Review Comment:
same as above
##########
core/src/test/java/org/apache/iceberg/TestFileScanTaskParser.java:
##########
@@ -27,29 +27,41 @@
import org.junit.jupiter.params.provider.ValueSource;
public class TestFileScanTaskParser {
+
Review Comment:
nit: unnecessary change
##########
core/src/test/java/org/apache/iceberg/TestDataTaskParser.java:
##########
@@ -0,0 +1,147 @@
+/*
+ * 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;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.assertj.core.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class TestDataTaskParser {
+ @Test
+ public void testStaticDataTask() {
+ StaticDataTask dataTask = (StaticDataTask) createDataTask();
+ String jsonStr = ScanTaskParser.toJson(dataTask);
+ Assertions.assertThat(jsonStr).isEqualTo(snapshotsDataTaskJson());
+ StaticDataTask deserializedTask = (StaticDataTask) ScanTaskParser.fromJson(jsonStr, true);
+ assertDataTaskEquals(dataTask, deserializedTask);
+ }
+
+ private DataTask createDataTask() {
+ Map<String, String> summary1 =
+ ImmutableMap.of(
+ "added-data-files", "1",
+ "added-records", "1",
+ "added-files-size", "10",
+ "changed-partition-count", "1",
+ "total-records", "1",
+ "total-files-size", "10",
+ "total-data-files", "1",
+ "total-delete-files", "0",
+ "total-position-deletes", "0",
+ "total-equality-deletes", "0");
+
+ Map<String, String> summary2 =
+ ImmutableMap.of(
+ "added-data-files", "1",
+ "added-records", "1",
+ "added-files-size", "10",
+ "changed-partition-count", "1",
+ "total-records", "2",
+ "total-files-size", "20",
+ "total-data-files", "2",
+ "total-delete-files", "0",
+ "total-position-deletes", "0",
+ "total-equality-deletes", "0");
+
+ List<Snapshot> snapshots =
+ Arrays.asList(
+ new BaseSnapshot(
+ 1L, 1L, null, 1234567890000L, "append", summary1, 1, "file:/tmp/manifest1.avro"),
+ new BaseSnapshot(
+ 2L, 2L, 1L, 9876543210000L, "append", summary2, 1, "file:/tmp/manifest2.avro"));
+
+ return StaticDataTask.of(
+ Files.localInput("file:/tmp/metadata2.json"),
+ SnapshotsTable.SNAPSHOT_SCHEMA,
+ SnapshotsTable.SNAPSHOT_SCHEMA,
+ snapshots,
+ SnapshotsTable::snapshotToRow);
+ }
+
+ private String snapshotsDataTaskJson() {
+ return "{\"task-type\":\"data-task\","
+ + "\"schema\":{\"type\":\"struct\",\"schema-id\":0,"
+ + "\"fields\":[{\"id\":1,\"name\":\"committed_at\",\"required\":true,\"type\":\"timestamptz\"},"
+ + "{\"id\":2,\"name\":\"snapshot_id\",\"required\":true,\"type\":\"long\"},"
+ + "{\"id\":3,\"name\":\"parent_id\",\"required\":false,\"type\":\"long\"},"
+ + "{\"id\":4,\"name\":\"operation\",\"required\":false,\"type\":\"string\"},"
+ + "{\"id\":5,\"name\":\"manifest_list\",\"required\":false,\"type\":\"string\"},"
+ + "{\"id\":6,\"name\":\"summary\",\"required\":false,\"type\":{\"type\":\"map\","
+ + "\"key-id\":7,\"key\":\"string\",\"value-id\":8,"
+ + "\"value\":\"string\",\"value-required\":true}}]},"
+ + "\"projection\":{\"type\":\"struct\",\"schema-id\":0,"
+ + "\"fields\":[{\"id\":1,\"name\":\"committed_at\",\"required\":true,\"type\":\"timestamptz\"},"
+ + "{\"id\":2,\"name\":\"snapshot_id\",\"required\":true,\"type\":\"long\"},"
+ + "{\"id\":3,\"name\":\"parent_id\",\"required\":false,\"type\":\"long\"},"
+ + "{\"id\":4,\"name\":\"operation\",\"required\":false,\"type\":\"string\"},"
+ + "{\"id\":5,\"name\":\"manifest_list\",\"required\":false,\"type\":\"string\"},"
+ + "{\"id\":6,\"name\":\"summary\",\"required\":false,\"type\":{\"type\":\"map\","
+ + "\"key-id\":7,\"key\":\"string\",\"value-id\":8,"
+ + "\"value\":\"string\",\"value-required\":true}}]},"
+ + "\"metadata-file\":{\"spec-id\":0,\"content\":\"DATA\","
+ + "\"file-path\":\"/tmp/metadata2.json\","
+ + "\"file-format\":\"METADATA\",\"partition\":{},"
+ + "\"file-size-in-bytes\":0,\"record-count\":2,\"sort-order-id\":0},"
+ + "\"rows\":[{\"1\":\"2009-02-13T23:31:30+00:00\",\"2\":1,\"4\":\"append\","
+ + "\"5\":\"file:/tmp/manifest1.avro\","
+ + "\"6\":{\"keys\":[\"added-data-files\",\"added-records\",\"added-files-size\",\"changed-partition-count\","
+ + "\"total-records\",\"total-files-size\",\"total-data-files\",\"total-delete-files\","
+ + "\"total-position-deletes\",\"total-equality-deletes\"],"
+ + "\"values\":[\"1\",\"1\",\"10\",\"1\",\"1\",\"10\",\"1\",\"0\",\"0\",\"0\"]}},"
+ + "{\"1\":\"2282-12-22T20:13:30+00:00\",\"2\":2,\"3\":1,\"4\":\"append\","
+ + "\"5\":\"file:/tmp/manifest2.avro\","
+ + "\"6\":{\"keys\":[\"added-data-files\",\"added-records\",\"added-files-size\",\"changed-partition-count\","
+ + "\"total-records\",\"total-files-size\",\"total-data-files\",\"total-delete-files\","
+ + "\"total-position-deletes\",\"total-equality-deletes\"],"
+ + "\"values\":[\"1\",\"1\",\"10\",\"1\",\"2\",\"20\",\"2\",\"0\",\"0\",\"0\"]}}]}";
+ }
+
+ private void assertDataTaskEquals(StaticDataTask expected, StaticDataTask actual) {
+ Assertions.assertThat(expected.schema().sameSchema(actual.schema()))
Review Comment:
```suggestion
assertThat(expected.schema().asStruct()).isEqualTo(atual.schema().asStruct()))
```
this is slightly better, because it will actually show the difference in the schema if the assertion ever fails, whereas `.isTrue()` will only say that the schema doesn't match
##########
core/src/test/java/org/apache/iceberg/TestDataTaskParser.java:
##########
@@ -0,0 +1,147 @@
+/*
+ * 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;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.assertj.core.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class TestDataTaskParser {
+ @Test
+ public void testStaticDataTask() {
+ StaticDataTask dataTask = (StaticDataTask) createDataTask();
+ String jsonStr = ScanTaskParser.toJson(dataTask);
+ Assertions.assertThat(jsonStr).isEqualTo(snapshotsDataTaskJson());
+ StaticDataTask deserializedTask = (StaticDataTask) ScanTaskParser.fromJson(jsonStr, true);
+ assertDataTaskEquals(dataTask, deserializedTask);
+ }
+
+ private DataTask createDataTask() {
+ Map<String, String> summary1 =
+ ImmutableMap.of(
+ "added-data-files", "1",
+ "added-records", "1",
+ "added-files-size", "10",
+ "changed-partition-count", "1",
+ "total-records", "1",
+ "total-files-size", "10",
+ "total-data-files", "1",
+ "total-delete-files", "0",
+ "total-position-deletes", "0",
+ "total-equality-deletes", "0");
+
+ Map<String, String> summary2 =
+ ImmutableMap.of(
+ "added-data-files", "1",
+ "added-records", "1",
+ "added-files-size", "10",
+ "changed-partition-count", "1",
+ "total-records", "2",
+ "total-files-size", "20",
+ "total-data-files", "2",
+ "total-delete-files", "0",
+ "total-position-deletes", "0",
+ "total-equality-deletes", "0");
+
+ List<Snapshot> snapshots =
+ Arrays.asList(
+ new BaseSnapshot(
+ 1L, 1L, null, 1234567890000L, "append", summary1, 1, "file:/tmp/manifest1.avro"),
+ new BaseSnapshot(
+ 2L, 2L, 1L, 9876543210000L, "append", summary2, 1, "file:/tmp/manifest2.avro"));
+
+ return StaticDataTask.of(
+ Files.localInput("file:/tmp/metadata2.json"),
+ SnapshotsTable.SNAPSHOT_SCHEMA,
+ SnapshotsTable.SNAPSHOT_SCHEMA,
+ snapshots,
+ SnapshotsTable::snapshotToRow);
+ }
+
+ private String snapshotsDataTaskJson() {
+ return "{\"task-type\":\"data-task\","
Review Comment:
it would be good to have this pretty-formatted so that it's easier to read
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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
Re: [PR] Core: add a new task-type field to task JSON serialization. add data task JSON serialization imp. [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #9728:
URL: https://github.com/apache/iceberg/pull/9728#discussion_r1526649078
##########
core/src/main/java/org/apache/iceberg/SnapshotsTable.java:
##########
@@ -27,7 +28,8 @@
* <p>This does not include snapshots that have been expired using {@link ExpireSnapshots}.
*/
public class SnapshotsTable extends BaseMetadataTable {
- private static final Schema SNAPSHOT_SCHEMA =
+ @VisibleForTesting
+ static final Schema SNAPSHOT_SCHEMA =
Review Comment:
I was testing a real `StaticDataTask` for snapshot metadata table/rows. Hence expose this as package private. Note that this is not `public` though.
We can avoid making this visible by define a custom test schema and custom static data task in the `TestDataTaskParser`.
let me know your thoughts/preference.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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
Re: [PR] Spec: add task-type field to JSON serialization of file scan task. add JSON serialization for StaticDataTask. [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #9728:
URL: https://github.com/apache/iceberg/pull/9728#discussion_r1490873382
##########
format/spec.md:
##########
@@ -1237,17 +1237,37 @@ Content file (data or delete) is serialized as a JSON object according to the fo
| **`equality-ids`** |`JSON list of int: Field ids used to determine row equality in equality delete files`|`[1]`|
| **`sort-order-id`** |`JSON int`|`1`|
-### File Scan Task Serialization
-
-File scan task is serialized as a JSON object according to the following table.
-
-| Metadata field |JSON representation|Example|
-|--------------------------|--- |--- |
-| **`schema`** |`JSON object`|`See above, read schemas instead`|
-| **`spec`** |`JSON object`|`See above, read partition specs instead`|
-| **`data-file`** |`JSON object`|`See above, read content file instead`|
-| **`delete-files`** |`JSON list of objects`|`See above, read content file instead`|
-| **`residual-filter`** |`JSON object: residual filter expression`|`{"type":"eq","term":"id","value":1}`|
+### Task Serialization
+
+There could be different implementations of tasks,
Review Comment:
```suggestion
There could be different implementations of tasks, e.g. `BaseFileScanTask` and `StaticDataTask` in Java.
```
this reads easier if it's on a single 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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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
Re: [PR] Spec, Core: add a new task-type field to task JSON serialization. add data task JSON serialization spec and imp. [iceberg]
Posted by "emkornfield (via GitHub)" <gi...@apache.org>.
emkornfield commented on code in PR #9728:
URL: https://github.com/apache/iceberg/pull/9728#discussion_r1492846630
##########
format/spec.md:
##########
@@ -1237,17 +1237,36 @@ Content file (data or delete) is serialized as a JSON object according to the fo
| **`equality-ids`** |`JSON list of int: Field ids used to determine row equality in equality delete files`|`[1]`|
| **`sort-order-id`** |`JSON int`|`1`|
-### File Scan Task Serialization
-
-File scan task is serialized as a JSON object according to the following table.
-
-| Metadata field |JSON representation|Example|
-|--------------------------|--- |--- |
-| **`schema`** |`JSON object`|`See above, read schemas instead`|
-| **`spec`** |`JSON object`|`See above, read partition specs instead`|
-| **`data-file`** |`JSON object`|`See above, read content file instead`|
-| **`delete-files`** |`JSON list of objects`|`See above, read content file instead`|
-| **`residual-filter`** |`JSON object: residual filter expression`|`{"type":"eq","term":"id","value":1}`|
+### Task Serialization
Review Comment:
Its really not clear to me why file scan task is added into the spec here? I don't think it is referenced any place in the main body of the specification?
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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
Re: [PR] Spec, Core: add a new task-type field to task JSON serialization. add data task JSON serialization spec and imp. [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #9728:
URL: https://github.com/apache/iceberg/pull/9728#discussion_r1493037575
##########
format/spec.md:
##########
@@ -1237,17 +1237,36 @@ Content file (data or delete) is serialized as a JSON object according to the fo
| **`equality-ids`** |`JSON list of int: Field ids used to determine row equality in equality delete files`|`[1]`|
| **`sort-order-id`** |`JSON int`|`1`|
-### File Scan Task Serialization
-
-File scan task is serialized as a JSON object according to the following table.
-
-| Metadata field |JSON representation|Example|
-|--------------------------|--- |--- |
-| **`schema`** |`JSON object`|`See above, read schemas instead`|
-| **`spec`** |`JSON object`|`See above, read partition specs instead`|
-| **`data-file`** |`JSON object`|`See above, read content file instead`|
-| **`delete-files`** |`JSON list of objects`|`See above, read content file instead`|
-| **`residual-filter`** |`JSON object: residual filter expression`|`{"type":"eq","term":"id","value":1}`|
+### Task Serialization
Review Comment:
@emkornfield that is a fair question. @rdblue also had a question if this should be added to REST OpenAPI. My questions is that this is also used by Flink for state serialization. Hence it is not just a REST thing.
what do others think? @rdblue @jackye1995 @aokolnychyi
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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