You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by yu...@apache.org on 2024/11/07 01:58:02 UTC
(paimon-python) branch release-0.1 updated (c973a50 -> cbb81bb)
This is an automated email from the ASF dual-hosted git repository.
yuzelin pushed a change to branch release-0.1
in repository https://gitbox.apache.org/repos/asf/paimon-python.git
from c973a50 Add release tools (#21)
new 46d44da [hotfix] Fix runtime exception on windows (#23)
new cbb81bb Read interface support predicate (#22)
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.github/workflows/paimon-python-checks.yml | 7 +
paimon_python_api/__init__.py | 5 +-
paimon_python_api/predicate.py | 95 +++++
paimon_python_api/read_builder.py | 13 +-
paimon_python_java/__init__.py | 7 +-
paimon_python_java/gateway_server.py | 3 +-
paimon_python_java/java_gateway.py | 1 +
.../org/apache/paimon/python/PredicationUtil.java | 111 ++++++
paimon_python_java/pypaimon.py | 103 +++++-
paimon_python_java/tests/test_preicates.py | 394 +++++++++++++++++++++
paimon_python_java/tests/utils.py | 5 +
11 files changed, 736 insertions(+), 8 deletions(-)
create mode 100644 paimon_python_api/predicate.py
create mode 100644 paimon_python_java/paimon-python-java-bridge/src/main/java/org/apache/paimon/python/PredicationUtil.java
create mode 100644 paimon_python_java/tests/test_preicates.py
(paimon-python) 02/02: Read interface support predicate (#22)
Posted by yu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yuzelin pushed a commit to branch release-0.1
in repository https://gitbox.apache.org/repos/asf/paimon-python.git
commit cbb81bbb66b86895fa18538fbeacfb0bd3dd758a
Author: yuzelin <33...@users.noreply.github.com>
AuthorDate: Wed Nov 6 17:49:17 2024 +0800
Read interface support predicate (#22)
(cherry picked from commit 5e7d4687e1106f4fb928fb9bbab1e50d15b99ff9)
---
.github/workflows/paimon-python-checks.yml | 7 +
paimon_python_api/__init__.py | 5 +-
paimon_python_api/predicate.py | 95 +++++
paimon_python_api/read_builder.py | 13 +-
paimon_python_java/__init__.py | 7 +-
paimon_python_java/gateway_server.py | 2 +-
paimon_python_java/java_gateway.py | 1 +
.../org/apache/paimon/python/PredicationUtil.java | 111 ++++++
paimon_python_java/pypaimon.py | 103 +++++-
paimon_python_java/tests/test_preicates.py | 394 +++++++++++++++++++++
paimon_python_java/tests/utils.py | 5 +
11 files changed, 735 insertions(+), 8 deletions(-)
diff --git a/.github/workflows/paimon-python-checks.yml b/.github/workflows/paimon-python-checks.yml
index e94820b..195783f 100644
--- a/.github/workflows/paimon-python-checks.yml
+++ b/.github/workflows/paimon-python-checks.yml
@@ -43,7 +43,14 @@ jobs:
with:
java-version: ${{ env.JDK_VERSION }}
distribution: 'adopt'
+ - name: Set up hadoop dependency
+ run: |
+ mkdir -p ${{ github.workspace }}/temp
+ curl -L -o ${{ github.workspace }}/temp/bundled-hadoop.jar \
+ https://repo.maven.apache.org/maven2/org/apache/flink/flink-shaded-hadoop-2-uber/2.8.3-10.0/flink-shaded-hadoop-2-uber-2.8.3-10.0.jar
- name: Run lint-python.sh
+ env:
+ _PYPAIMON_HADOOP_CLASSPATH: ${{ github.workspace }}/temp/bundled-hadoop.jar
run: |
chmod +x dev/lint-python.sh
./dev/lint-python.sh
diff --git a/paimon_python_api/__init__.py b/paimon_python_api/__init__.py
index 86090c9..44717bf 100644
--- a/paimon_python_api/__init__.py
+++ b/paimon_python_api/__init__.py
@@ -19,6 +19,7 @@
from .split import Split
from .table_read import TableRead
from .table_scan import TableScan, Plan
+from .predicate import Predicate, PredicateBuilder
from .read_builder import ReadBuilder
from .commit_message import CommitMessage
from .table_commit import BatchTableCommit
@@ -39,5 +40,7 @@ __all__ = [
'BatchWriteBuilder',
'Table',
'Schema',
- 'Catalog'
+ 'Catalog',
+ 'Predicate',
+ 'PredicateBuilder'
]
diff --git a/paimon_python_api/predicate.py b/paimon_python_api/predicate.py
new file mode 100644
index 0000000..46280d1
--- /dev/null
+++ b/paimon_python_api/predicate.py
@@ -0,0 +1,95 @@
+################################################################################
+# 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.
+#################################################################################
+
+from abc import ABC, abstractmethod
+from typing import Any, List
+
+
+class Predicate(ABC):
+ """Predicate which evaluates to a boolean. Now it doesn't have
+ any methods because only paimon_python_java implement it and
+ the Java implementation convert it to Java object."""
+
+
+class PredicateBuilder(ABC):
+ """A utility class to create Predicate object for common filter conditions."""
+
+ @abstractmethod
+ def equal(self, field: str, literal: Any) -> Predicate:
+ """field = literal"""
+
+ @abstractmethod
+ def not_equal(self, field: str, literal: Any) -> Predicate:
+ """field <> literal"""
+
+ @abstractmethod
+ def less_than(self, field: str, literal: Any) -> Predicate:
+ """field < literal"""
+
+ @abstractmethod
+ def less_or_equal(self, field: str, literal: Any) -> Predicate:
+ """field <= literal"""
+
+ @abstractmethod
+ def greater_than(self, field: str, literal: Any) -> Predicate:
+ """field > literal"""
+
+ @abstractmethod
+ def greater_or_equal(self, field: str, literal: Any) -> Predicate:
+ """field >= literal"""
+
+ @abstractmethod
+ def is_null(self, field: str) -> Predicate:
+ """field IS NULL"""
+
+ @abstractmethod
+ def is_not_null(self, field: str) -> Predicate:
+ """field IS NOT NULL"""
+
+ @abstractmethod
+ def startswith(self, field: str, pattern_literal: Any) -> Predicate:
+ """field.startswith"""
+
+ @abstractmethod
+ def endswith(self, field: str, pattern_literal: Any) -> Predicate:
+ """field.endswith()"""
+
+ @abstractmethod
+ def contains(self, field: str, pattern_literal: Any) -> Predicate:
+ """literal in field"""
+
+ @abstractmethod
+ def is_in(self, field: str, literals: List[Any]) -> Predicate:
+ """field IN literals"""
+
+ @abstractmethod
+ def is_not_in(self, field: str, literals: List[Any]) -> Predicate:
+ """field NOT IN literals"""
+
+ @abstractmethod
+ def between(self, field: str, included_lower_bound: Any, included_upper_bound: Any) \
+ -> Predicate:
+ """field BETWEEN included_lower_bound AND included_upper_bound"""
+
+ @abstractmethod
+ def and_predicates(self, predicates: List[Predicate]) -> Predicate:
+ """predicate1 AND predicate2 AND ..."""
+
+ @abstractmethod
+ def or_predicates(self, predicates: List[Predicate]) -> Predicate:
+ """predicate1 OR predicate2 OR ..."""
diff --git a/paimon_python_api/read_builder.py b/paimon_python_api/read_builder.py
index 94ec073..ad5e6d6 100644
--- a/paimon_python_api/read_builder.py
+++ b/paimon_python_api/read_builder.py
@@ -17,13 +17,20 @@
#################################################################################
from abc import ABC, abstractmethod
-from paimon_python_api import TableRead, TableScan
+from paimon_python_api import TableRead, TableScan, Predicate, PredicateBuilder
from typing import List
class ReadBuilder(ABC):
"""An interface for building the TableScan and TableRead."""
+ @abstractmethod
+ def with_filter(self, predicate: Predicate):
+ """
+ Push filters, will filter the data as much as possible,
+ but it is not guaranteed that it is a complete filter.
+ """
+
@abstractmethod
def with_projection(self, projection: List[List[int]]) -> 'ReadBuilder':
"""Push nested projection."""
@@ -39,3 +46,7 @@ class ReadBuilder(ABC):
@abstractmethod
def new_read(self) -> TableRead:
"""Create a TableRead to read splits."""
+
+ @abstractmethod
+ def new_predicate_builder(self) -> PredicateBuilder:
+ """Create a builder for Predicate."""
diff --git a/paimon_python_java/__init__.py b/paimon_python_java/__init__.py
index 6e97d9e..9b0d002 100644
--- a/paimon_python_java/__init__.py
+++ b/paimon_python_java/__init__.py
@@ -18,7 +18,8 @@
from .util import constants
from .pypaimon import (Catalog, Table, ReadBuilder, TableScan, Plan, Split, TableRead,
- BatchWriteBuilder, BatchTableWrite, CommitMessage, BatchTableCommit)
+ BatchWriteBuilder, BatchTableWrite, CommitMessage, BatchTableCommit,
+ Predicate, PredicateBuilder)
__all__ = [
'constants',
@@ -32,5 +33,7 @@ __all__ = [
'BatchWriteBuilder',
'BatchTableWrite',
'CommitMessage',
- 'BatchTableCommit'
+ 'BatchTableCommit',
+ 'Predicate',
+ 'PredicateBuilder'
]
diff --git a/paimon_python_java/gateway_server.py b/paimon_python_java/gateway_server.py
index 7285e98..2061d59 100644
--- a/paimon_python_java/gateway_server.py
+++ b/paimon_python_java/gateway_server.py
@@ -103,7 +103,7 @@ def _get_hadoop_classpath(env):
return env[constants.PYPAIMON_HADOOP_CLASSPATH]
if 'HADOOP_CLASSPATH' in env:
- return None
+ return env['HADOOP_CLASSPATH']
else:
raise EnvironmentError(f"You haven't set '{constants.PYPAIMON_HADOOP_CLASSPATH}', \
and 'HADOOP_CLASSPATH' is also not set. Ensure one of them is set.")
diff --git a/paimon_python_java/java_gateway.py b/paimon_python_java/java_gateway.py
index f2b1621..3dabcfd 100644
--- a/paimon_python_java/java_gateway.py
+++ b/paimon_python_java/java_gateway.py
@@ -109,6 +109,7 @@ def import_paimon_view(gateway):
java_import(gateway.jvm, 'org.apache.paimon.types.*')
java_import(gateway.jvm, 'org.apache.paimon.python.*')
java_import(gateway.jvm, "org.apache.paimon.data.*")
+ java_import(gateway.jvm, "org.apache.paimon.predicate.PredicateBuilder")
class Watchdog(object):
diff --git a/paimon_python_java/paimon-python-java-bridge/src/main/java/org/apache/paimon/python/PredicationUtil.java b/paimon_python_java/paimon-python-java-bridge/src/main/java/org/apache/paimon/python/PredicationUtil.java
new file mode 100644
index 0000000..a863dfd
--- /dev/null
+++ b/paimon_python_java/paimon-python-java-bridge/src/main/java/org/apache/paimon/python/PredicationUtil.java
@@ -0,0 +1,111 @@
+/*
+ * 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.paimon.python;
+
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.predicate.PredicateBuilder;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.RowType;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+/** For building Predicate. */
+public class PredicationUtil {
+
+ public static Predicate build(
+ RowType rowType,
+ PredicateBuilder builder,
+ String method,
+ int index,
+ List<Object> literals) {
+ literals =
+ literals.stream()
+ .map(l -> convertJavaObject(rowType.getTypeAt(index), l))
+ .collect(Collectors.toList());
+ switch (method) {
+ case "equal":
+ return builder.equal(index, literals.get(0));
+ case "notEqual":
+ return builder.notEqual(index, literals.get(0));
+ case "lessThan":
+ return builder.lessThan(index, literals.get(0));
+ case "lessOrEqual":
+ return builder.lessOrEqual(index, literals.get(0));
+ case "greaterThan":
+ return builder.greaterThan(index, literals.get(0));
+ case "greaterOrEqual":
+ return builder.greaterOrEqual(index, literals.get(0));
+ case "isNull":
+ return builder.isNull(index);
+ case "isNotNull":
+ return builder.isNotNull(index);
+ case "startsWith":
+ return builder.startsWith(index, literals.get(0));
+ case "endsWith":
+ return builder.endsWith(index, literals.get(0));
+ case "contains":
+ return builder.contains(index, literals.get(0));
+ case "in":
+ return builder.in(index, literals);
+ case "notIn":
+ return builder.notIn(index, literals);
+ case "between":
+ return builder.between(index, literals.get(0), literals.get(1));
+ default:
+ throw new UnsupportedOperationException(
+ "Unknown PredicateBuilder method " + method);
+ }
+ }
+
+ /** Some type is not convenient to transfer from Python to Java. */
+ private static Object convertJavaObject(DataType literalType, Object literal) {
+ switch (literalType.getTypeRoot()) {
+ case BOOLEAN:
+ case DOUBLE:
+ case INTEGER:
+ return literal;
+ case CHAR:
+ case VARCHAR:
+ return BinaryString.fromString((String) literal);
+ case FLOAT:
+ return ((Number) literal).floatValue();
+ case TINYINT:
+ return ((Number) literal).byteValue();
+ case SMALLINT:
+ return ((Number) literal).shortValue();
+ case BIGINT:
+ return ((Number) literal).longValue();
+ default:
+ throw new UnsupportedOperationException(
+ "Unsupported predicate leaf type " + literalType.getTypeRoot().name());
+ }
+ }
+
+ public static Predicate buildAnd(List<Predicate> predicates) {
+ // 'and' is keyword of Python
+ return PredicateBuilder.and(predicates);
+ }
+
+ public static Predicate buildOr(List<Predicate> predicates) {
+ // 'or' is keyword of Python
+ return PredicateBuilder.or(predicates);
+ }
+}
diff --git a/paimon_python_java/pypaimon.py b/paimon_python_java/pypaimon.py
index fcf0695..0d3101b 100644
--- a/paimon_python_java/pypaimon.py
+++ b/paimon_python_java/pypaimon.py
@@ -22,8 +22,9 @@ import pyarrow as pa
from paimon_python_java.java_gateway import get_gateway
from paimon_python_java.util import java_utils, constants
from paimon_python_api import (catalog, table, read_builder, table_scan, split, table_read,
- write_builder, table_write, commit_message, table_commit, Schema)
-from typing import List, Iterator, Optional
+ write_builder, table_write, commit_message, table_commit, Schema,
+ predicate)
+from typing import List, Iterator, Optional, Any
class Catalog(catalog.Catalog):
@@ -85,6 +86,10 @@ class ReadBuilder(read_builder.ReadBuilder):
self._catalog_options = catalog_options
self._arrow_schema = arrow_schema
+ def with_filter(self, predicate: 'Predicate'):
+ self._j_read_builder.withFilter(predicate.to_j_predicate())
+ return self
+
def with_projection(self, projection: List[List[int]]) -> 'ReadBuilder':
self._j_read_builder.withProjection(projection)
return self
@@ -98,9 +103,12 @@ class ReadBuilder(read_builder.ReadBuilder):
return TableScan(j_table_scan)
def new_read(self) -> 'TableRead':
- j_table_read = self._j_read_builder.newRead()
+ j_table_read = self._j_read_builder.newRead().executeFilter()
return TableRead(j_table_read, self._j_row_type, self._catalog_options, self._arrow_schema)
+ def new_predicate_builder(self) -> 'PredicateBuilder':
+ return PredicateBuilder(self._j_row_type)
+
class TableScan(table_scan.TableScan):
@@ -257,3 +265,92 @@ class BatchTableCommit(table_commit.BatchTableCommit):
def close(self):
self._j_batch_table_commit.close()
+
+
+class Predicate(predicate.Predicate):
+
+ def __init__(self, j_predicate):
+ self._j_predicate = j_predicate
+
+ def to_j_predicate(self):
+ return self._j_predicate
+
+
+class PredicateBuilder(predicate.PredicateBuilder):
+
+ def __init__(self, j_row_type):
+ self._field_names = j_row_type.getFieldNames()
+ self._j_row_type = j_row_type
+ self._j_predicate_builder = get_gateway().jvm.PredicateBuilder(j_row_type)
+
+ def _build(self, method: str, field: str, literals: Optional[List[Any]] = None):
+ error = ValueError(f'The field {field} is not in field list {self._field_names}.')
+ try:
+ index = self._field_names.index(field)
+ if index == -1:
+ raise error
+ except ValueError:
+ raise error
+
+ if literals is None:
+ literals = []
+
+ j_predicate = get_gateway().jvm.PredicationUtil.build(
+ self._j_row_type,
+ self._j_predicate_builder,
+ method,
+ index,
+ literals
+ )
+ return Predicate(j_predicate)
+
+ def equal(self, field: str, literal: Any) -> Predicate:
+ return self._build('equal', field, [literal])
+
+ def not_equal(self, field: str, literal: Any) -> Predicate:
+ return self._build('notEqual', field, [literal])
+
+ def less_than(self, field: str, literal: Any) -> Predicate:
+ return self._build('lessThan', field, [literal])
+
+ def less_or_equal(self, field: str, literal: Any) -> Predicate:
+ return self._build('lessOrEqual', field, [literal])
+
+ def greater_than(self, field: str, literal: Any) -> Predicate:
+ return self._build('greaterThan', field, [literal])
+
+ def greater_or_equal(self, field: str, literal: Any) -> Predicate:
+ return self._build('greaterOrEqual', field, [literal])
+
+ def is_null(self, field: str) -> Predicate:
+ return self._build('isNull', field)
+
+ def is_not_null(self, field: str) -> Predicate:
+ return self._build('isNotNull', field)
+
+ def startswith(self, field: str, pattern_literal: Any) -> Predicate:
+ return self._build('startsWith', field, [pattern_literal])
+
+ def endswith(self, field: str, pattern_literal: Any) -> Predicate:
+ return self._build('endsWith', field, [pattern_literal])
+
+ def contains(self, field: str, pattern_literal: Any) -> Predicate:
+ return self._build('contains', field, [pattern_literal])
+
+ def is_in(self, field: str, literals: List[Any]) -> Predicate:
+ return self._build('in', field, literals)
+
+ def is_not_in(self, field: str, literals: List[Any]) -> Predicate:
+ return self._build('notIn', field, literals)
+
+ def between(self, field: str, included_lower_bound: Any, included_upper_bound: Any) \
+ -> Predicate:
+ return self._build('between', field, [included_lower_bound, included_upper_bound])
+
+ def and_predicates(self, predicates: List[Predicate]) -> Predicate:
+ predicates = list(map(lambda p: p.to_j_predicate(), predicates))
+ return Predicate(get_gateway().jvm.PredicationUtil.buildAnd(predicates))
+
+ def or_predicates(self, predicates: List[Predicate]) -> Predicate:
+ predicates = list(map(lambda p: p.to_j_predicate(), predicates))
+ return Predicate(get_gateway().jvm.PredicationUtil.buildOr(predicates))
diff --git a/paimon_python_java/tests/test_preicates.py b/paimon_python_java/tests/test_preicates.py
new file mode 100644
index 0000000..7ee1a91
--- /dev/null
+++ b/paimon_python_java/tests/test_preicates.py
@@ -0,0 +1,394 @@
+################################################################################
+# 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.
+################################################################################
+
+import os
+import shutil
+import tempfile
+import unittest
+import random
+import pandas as pd
+import pyarrow as pa
+
+from paimon_python_api import Schema
+from paimon_python_java import Catalog
+from paimon_python_java.tests import utils
+from setup_utils import java_setuputils
+
+
+def _check_filtered_result(read_builder, expected_df):
+ scan = read_builder.new_scan()
+ read = read_builder.new_read()
+ actual_df = read.to_pandas(scan.plan().splits())
+ pd.testing.assert_frame_equal(
+ actual_df.reset_index(drop=True), expected_df.reset_index(drop=True))
+
+
+# TODO: parquet has bug now
+def _random_format():
+ return random.choice(['avro', 'orc'])
+
+
+class PredicateTest(unittest.TestCase):
+
+ @classmethod
+ def setUpClass(cls):
+ java_setuputils.setup_java_bridge()
+ cls.hadoop_path = tempfile.mkdtemp()
+ utils.setup_hadoop_bundle_jar(cls.hadoop_path)
+ cls.warehouse = tempfile.mkdtemp()
+
+ catalog = Catalog.create({'warehouse': cls.warehouse})
+ catalog.create_database('default', False)
+
+ pa_schema = pa.schema([
+ ('f0', pa.int64()),
+ ('f1', pa.string()),
+ ])
+ catalog.create_table('default.test_append',
+ Schema(pa_schema, options={'file.format': _random_format()}),
+ False)
+ catalog.create_table('default.test_pk',
+ Schema(pa_schema, primary_keys=['f0'],
+ options={'bucket': '1', 'file.format': _random_format()}),
+ False)
+
+ df = pd.DataFrame({
+ 'f0': [1, 2, 3, 4, 5],
+ 'f1': ['abc', 'abbc', 'bc', 'd', None],
+ })
+
+ append_table = catalog.get_table('default.test_append')
+ write_builder = append_table.new_batch_write_builder()
+ write = write_builder.new_write()
+ commit = write_builder.new_commit()
+ write.write_pandas(df)
+ commit.commit(write.prepare_commit())
+ write.close()
+ commit.close()
+
+ pk_table = catalog.get_table('default.test_pk')
+ write_builder = pk_table.new_batch_write_builder()
+ write = write_builder.new_write()
+ commit = write_builder.new_commit()
+ write.write_pandas(df)
+ commit.commit(write.prepare_commit())
+ write.close()
+ commit.close()
+
+ cls.catalog = catalog
+ cls.df = df
+
+ @classmethod
+ def tearDownClass(cls):
+ java_setuputils.clean()
+ if os.path.exists(cls.hadoop_path):
+ shutil.rmtree(cls.hadoop_path)
+ if os.path.exists(cls.warehouse):
+ shutil.rmtree(cls.warehouse)
+
+ def testWrongFieldName(self):
+ table = self.catalog.get_table('default.test_append')
+ predicate_builder = table.new_read_builder().new_predicate_builder()
+ with self.assertRaises(ValueError) as e:
+ predicate_builder.equal('f2', 'a')
+ self.assertEqual(str(e.exception), "The field f2 is not in field list ['f0', 'f1'].")
+
+ def testAppendWithDuplicate(self):
+ pa_schema = pa.schema([
+ ('f0', pa.int64()),
+ ('f1', pa.string()),
+ ])
+ self.catalog.create_table('default.test_append_with_duplicate', Schema(pa_schema), False)
+
+ df = pd.DataFrame({
+ 'f0': [1, 1, 2, 2],
+ 'f1': ['a', 'b', 'c', 'd'],
+ })
+
+ table = self.catalog.get_table('default.test_append_with_duplicate')
+ write_builder = table.new_batch_write_builder()
+ write = write_builder.new_write()
+ commit = write_builder.new_commit()
+ write.write_pandas(df)
+ commit.commit(write.prepare_commit())
+ write.close()
+ commit.close()
+
+ predicate_builder = table.new_read_builder().new_predicate_builder()
+
+ predicate = predicate_builder.equal('f0', 1)
+ _check_filtered_result(table.new_read_builder().with_filter(predicate), df.loc[0:1])
+
+ predicate = predicate_builder.equal('f0', 0)
+ read_builder = table.new_read_builder().with_filter(predicate)
+ scan = read_builder.new_scan()
+ read = read_builder.new_read()
+ actual_df = read.to_pandas(scan.plan().splits())
+ self.assertEqual(len(actual_df), 0)
+
+ def testAllFieldTypesWithEqual(self):
+ pa_schema = pa.schema([
+ # int
+ ('_tinyint', pa.int8()),
+ ('_smallint', pa.int16()),
+ ('_int', pa.int32()),
+ ('_bigint', pa.int64()),
+ # float
+ ('_float16', pa.float32()), # NOTE: cannot write pa.float16() data into Paimon
+ ('_float32', pa.float32()),
+ ('_double', pa.float64()),
+ # string
+ ('_string', pa.string()),
+ # bool
+ ('_boolean', pa.bool_())
+ ])
+ self.catalog.create_table('default.test_all_field_types',
+ Schema(pa_schema, options={'file.format': _random_format()}),
+ False)
+ table = self.catalog.get_table('default.test_all_field_types')
+ write_builder = table.new_batch_write_builder()
+ write = write_builder.new_write()
+ commit = write_builder.new_commit()
+
+ df = pd.DataFrame({
+ '_tinyint': pd.Series([1, 2], dtype='int8'),
+ '_smallint': pd.Series([10, 20], dtype='int16'),
+ '_int': pd.Series([100, 200], dtype='int32'),
+ '_bigint': pd.Series([1000, 2000], dtype='int64'),
+ '_float16': pd.Series([1.0, 2.0], dtype='float16'),
+ '_float32': pd.Series([1.00, 2.00], dtype='float32'),
+ '_double': pd.Series([1.000, 2.000], dtype='double'),
+ '_string': pd.Series(['A', 'B'], dtype='object'),
+ '_boolean': [True, False]
+ })
+ record_batch = pa.RecordBatch.from_pandas(df, schema=pa_schema)
+ # prepare for assertion
+ df['_float16'] = df['_float16'].astype('float32')
+
+ write.write_arrow_batch(record_batch)
+ commit.commit(write.prepare_commit())
+ write.close()
+ commit.close()
+
+ predicate_builder = table.new_read_builder().new_predicate_builder()
+
+ predicate = predicate_builder.equal('_tinyint', 1)
+ _check_filtered_result(table.new_read_builder().with_filter(predicate), df.loc[[0]])
+
+ predicate = predicate_builder.equal('_smallint', 20)
+ _check_filtered_result(table.new_read_builder().with_filter(predicate), df.loc[[1]])
+
+ predicate = predicate_builder.equal('_int', 100)
+ _check_filtered_result(table.new_read_builder().with_filter(predicate), df.loc[[0]])
+
+ predicate = predicate_builder.equal('_bigint', 2000)
+ _check_filtered_result(table.new_read_builder().with_filter(predicate), df.loc[[1]])
+
+ predicate = predicate_builder.equal('_float16', 1.0)
+ _check_filtered_result(table.new_read_builder().with_filter(predicate), df.loc[[0]])
+
+ predicate = predicate_builder.equal('_float32', 2.00)
+ _check_filtered_result(table.new_read_builder().with_filter(predicate), df.loc[[1]])
+
+ predicate = predicate_builder.equal('_double', 1.000)
+ _check_filtered_result(table.new_read_builder().with_filter(predicate), df.loc[[0]])
+
+ predicate = predicate_builder.equal('_string', 'B')
+ _check_filtered_result(table.new_read_builder().with_filter(predicate), df.loc[[1]])
+
+ predicate = predicate_builder.equal('_boolean', True)
+ _check_filtered_result(table.new_read_builder().with_filter(predicate), df.loc[[0]])
+
+ def testEqualPk(self):
+ table = self.catalog.get_table('default.test_pk')
+ predicate_builder = table.new_read_builder().new_predicate_builder()
+ predicate = predicate_builder.equal('f0', 1)
+ _check_filtered_result(table.new_read_builder().with_filter(predicate), self.df.loc[[0]])
+
+ def testNotEqualAppend(self):
+ table = self.catalog.get_table('default.test_append')
+ predicate_builder = table.new_read_builder().new_predicate_builder()
+ predicate = predicate_builder.not_equal('f0', 1)
+ _check_filtered_result(table.new_read_builder().with_filter(predicate), self.df.loc[1:4])
+
+ def testNotEqualPk(self):
+ table = self.catalog.get_table('default.test_pk')
+ predicate_builder = table.new_read_builder().new_predicate_builder()
+ predicate = predicate_builder.not_equal('f0', 1)
+ _check_filtered_result(table.new_read_builder().with_filter(predicate), self.df.loc[1:4])
+
+ def testLessThanAppend(self):
+ table = self.catalog.get_table('default.test_append')
+ predicate_builder = table.new_read_builder().new_predicate_builder()
+ predicate = predicate_builder.less_than('f0', 3)
+ _check_filtered_result(table.new_read_builder().with_filter(predicate), self.df.loc[0:1])
+
+ def testLessThanPk(self):
+ table = self.catalog.get_table('default.test_pk')
+ predicate_builder = table.new_read_builder().new_predicate_builder()
+ predicate = predicate_builder.less_than('f0', 3)
+ _check_filtered_result(table.new_read_builder().with_filter(predicate), self.df.loc[0:1])
+
+ def testLessOrEqualAppend(self):
+ table = self.catalog.get_table('default.test_append')
+ predicate_builder = table.new_read_builder().new_predicate_builder()
+ predicate = predicate_builder.less_or_equal('f0', 3)
+ _check_filtered_result(table.new_read_builder().with_filter(predicate), self.df.loc[0:2])
+
+ def testLessOrEqualPk(self):
+ table = self.catalog.get_table('default.test_pk')
+ predicate_builder = table.new_read_builder().new_predicate_builder()
+ predicate = predicate_builder.less_or_equal('f0', 3)
+ _check_filtered_result(table.new_read_builder().with_filter(predicate), self.df.loc[0:2])
+
+ def testGreaterThanAppend(self):
+ table = self.catalog.get_table('default.test_append')
+ predicate_builder = table.new_read_builder().new_predicate_builder()
+ predicate = predicate_builder.greater_than('f0', 3)
+ _check_filtered_result(table.new_read_builder().with_filter(predicate), self.df.loc[3:4])
+
+ def testGreaterThanPk(self):
+ table = self.catalog.get_table('default.test_pk')
+ predicate_builder = table.new_read_builder().new_predicate_builder()
+ predicate = predicate_builder.greater_than('f0', 3)
+ _check_filtered_result(table.new_read_builder().with_filter(predicate), self.df.loc[3:4])
+
+ def testGreaterOrEqualAppend(self):
+ table = self.catalog.get_table('default.test_append')
+ predicate_builder = table.new_read_builder().new_predicate_builder()
+ predicate = predicate_builder.greater_or_equal('f0', 3)
+ _check_filtered_result(table.new_read_builder().with_filter(predicate), self.df.loc[2:4])
+
+ def testGreaterOrEqualPk(self):
+ table = self.catalog.get_table('default.test_pk')
+ predicate_builder = table.new_read_builder().new_predicate_builder()
+ predicate = predicate_builder.greater_or_equal('f0', 3)
+ _check_filtered_result(table.new_read_builder().with_filter(predicate), self.df.loc[2:4])
+
+ def testIsNullAppend(self):
+ table = self.catalog.get_table('default.test_append')
+ predicate_builder = table.new_read_builder().new_predicate_builder()
+ predicate = predicate_builder.is_null('f1')
+ _check_filtered_result(table.new_read_builder().with_filter(predicate), self.df.loc[[4]])
+
+ def testIsNullPk(self):
+ table = self.catalog.get_table('default.test_pk')
+ predicate_builder = table.new_read_builder().new_predicate_builder()
+ predicate = predicate_builder.is_null('f1')
+ _check_filtered_result(table.new_read_builder().with_filter(predicate), self.df.loc[[4]])
+
+ def testIsNotNullAppend(self):
+ table = self.catalog.get_table('default.test_append')
+ predicate_builder = table.new_read_builder().new_predicate_builder()
+ predicate = predicate_builder.is_not_null('f1')
+ _check_filtered_result(table.new_read_builder().with_filter(predicate), self.df.loc[0:3])
+
+ def testIsNotNullPk(self):
+ table = self.catalog.get_table('default.test_pk')
+ predicate_builder = table.new_read_builder().new_predicate_builder()
+ predicate = predicate_builder.is_not_null('f1')
+ _check_filtered_result(table.new_read_builder().with_filter(predicate), self.df.loc[0:3])
+
+ def testStartswithAppend(self):
+ table = self.catalog.get_table('default.test_append')
+ predicate_builder = table.new_read_builder().new_predicate_builder()
+ predicate = predicate_builder.startswith('f1', 'ab')
+ _check_filtered_result(table.new_read_builder().with_filter(predicate), self.df.loc[0:1])
+
+ def testStartswithPk(self):
+ table = self.catalog.get_table('default.test_pk')
+ predicate_builder = table.new_read_builder().new_predicate_builder()
+ predicate = predicate_builder.startswith('f1', 'ab')
+ _check_filtered_result(table.new_read_builder().with_filter(predicate), self.df.loc[0:1])
+
+ def testEndswithAppend(self):
+ table = self.catalog.get_table('default.test_append')
+ predicate_builder = table.new_read_builder().new_predicate_builder()
+ predicate = predicate_builder.endswith('f1', 'bc')
+ _check_filtered_result(table.new_read_builder().with_filter(predicate), self.df.loc[0:2])
+
+ def testEndswithPk(self):
+ table = self.catalog.get_table('default.test_pk')
+ predicate_builder = table.new_read_builder().new_predicate_builder()
+ predicate = predicate_builder.endswith('f1', 'bc')
+ _check_filtered_result(table.new_read_builder().with_filter(predicate), self.df.loc[0:2])
+
+ def testContainsAppend(self):
+ table = self.catalog.get_table('default.test_append')
+ predicate_builder = table.new_read_builder().new_predicate_builder()
+ predicate = predicate_builder.contains('f1', 'bb')
+ _check_filtered_result(table.new_read_builder().with_filter(predicate), self.df.loc[[1]])
+
+ def testContainsPk(self):
+ table = self.catalog.get_table('default.test_pk')
+ predicate_builder = table.new_read_builder().new_predicate_builder()
+ predicate = predicate_builder.contains('f1', 'bb')
+ _check_filtered_result(table.new_read_builder().with_filter(predicate), self.df.loc[[1]])
+
+ def testIsInAppend(self):
+ table = self.catalog.get_table('default.test_append')
+ predicate_builder = table.new_read_builder().new_predicate_builder()
+ predicate = predicate_builder.is_in('f0', [1, 2])
+ _check_filtered_result(table.new_read_builder().with_filter(predicate), self.df.loc[0:1])
+
+ def testIsInPk(self):
+ table = self.catalog.get_table('default.test_pk')
+ predicate_builder = table.new_read_builder().new_predicate_builder()
+ predicate = predicate_builder.is_in('f1', ['abc', 'd'])
+ _check_filtered_result(table.new_read_builder().with_filter(predicate), self.df.loc[[0, 3]])
+
+ def testIsNotInAppend(self):
+ table = self.catalog.get_table('default.test_append')
+ predicate_builder = table.new_read_builder().new_predicate_builder()
+ predicate = predicate_builder.is_not_in('f0', [1, 2])
+ _check_filtered_result(table.new_read_builder().with_filter(predicate), self.df.loc[2:4])
+
+ def testIsNotInPk(self):
+ table = self.catalog.get_table('default.test_pk')
+ predicate_builder = table.new_read_builder().new_predicate_builder()
+ predicate = predicate_builder.is_not_in('f1', ['abc', 'd'])
+ _check_filtered_result(table.new_read_builder().with_filter(predicate), self.df.loc[1:2])
+
+ def testBetweenAppend(self):
+ table = self.catalog.get_table('default.test_append')
+ predicate_builder = table.new_read_builder().new_predicate_builder()
+ predicate = predicate_builder.between('f0', 1, 3)
+ _check_filtered_result(table.new_read_builder().with_filter(predicate), self.df.loc[0:2])
+
+ def testBetweenPk(self):
+ table = self.catalog.get_table('default.test_pk')
+ predicate_builder = table.new_read_builder().new_predicate_builder()
+ predicate = predicate_builder.between('f0', 1, 3)
+ _check_filtered_result(table.new_read_builder().with_filter(predicate), self.df.loc[0:2])
+
+ def testAndPredicates(self):
+ table = self.catalog.get_table('default.test_append')
+ predicate_builder = table.new_read_builder().new_predicate_builder()
+ predicate1 = predicate_builder.greater_than('f0', 1)
+ predicate2 = predicate_builder.startswith('f1', 'ab')
+ predicate = predicate_builder.and_predicates([predicate1, predicate2])
+ _check_filtered_result(table.new_read_builder().with_filter(predicate), self.df.loc[[1]])
+
+ def testOrPredicates(self):
+ table = self.catalog.get_table('default.test_append')
+ predicate_builder = table.new_read_builder().new_predicate_builder()
+ predicate1 = predicate_builder.greater_than('f0', 3)
+ predicate2 = predicate_builder.less_than('f0', 2)
+ predicate = predicate_builder.or_predicates([predicate1, predicate2])
+ _check_filtered_result(table.new_read_builder().with_filter(predicate),
+ self.df.loc[[0, 3, 4]])
diff --git a/paimon_python_java/tests/utils.py b/paimon_python_java/tests/utils.py
index 2af2fe0..350f80e 100644
--- a/paimon_python_java/tests/utils.py
+++ b/paimon_python_java/tests/utils.py
@@ -23,6 +23,11 @@ from paimon_python_java.util import constants
def setup_hadoop_bundle_jar(hadoop_dir):
+ if constants.PYPAIMON_HADOOP_CLASSPATH in os.environ:
+ file = os.environ[constants.PYPAIMON_HADOOP_CLASSPATH]
+ if os.path.isfile(file):
+ return
+
url = 'https://repo.maven.apache.org/maven2/org/apache/flink/' \
'flink-shaded-hadoop-2-uber/2.8.3-10.0/flink-shaded-hadoop-2-uber-2.8.3-10.0.jar'
(paimon-python) 01/02: [hotfix] Fix runtime exception on windows (#23)
Posted by yu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yuzelin pushed a commit to branch release-0.1
in repository https://gitbox.apache.org/repos/asf/paimon-python.git
commit 46d44dae75d53f07b4837ab439e935da75552dc4
Author: yuzelin <33...@users.noreply.github.com>
AuthorDate: Thu Oct 31 17:21:33 2024 +0800
[hotfix] Fix runtime exception on windows (#23)
(cherry picked from commit 0d0237b9e1d023f57f958dcbb83cdbf77aac8e5b)
---
paimon_python_java/gateway_server.py | 1 +
1 file changed, 1 insertion(+)
diff --git a/paimon_python_java/gateway_server.py b/paimon_python_java/gateway_server.py
index fc05ddc..7285e98 100644
--- a/paimon_python_java/gateway_server.py
+++ b/paimon_python_java/gateway_server.py
@@ -63,6 +63,7 @@ def launch_gateway_server_process(env):
*main_args
]
+ preexec_fn = None
if not on_windows():
def preexec_func():
# ignore ctrl-c / SIGINT