T

Author: yuzelin

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.")

paimon_python_java/tests/test_preicates.py

@@ -54,44 +54,52 @@ def setUpClass(cls):
 
         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)
+        cls.catalog = catalog
 
         df = pd.DataFrame({
             'f0': [1, 2, 3, 4, 5],
             'f1': ['abc', 'abbc', 'bc', 'd', None],
         })
+        cls.df = df
 
-        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()
+    def _init(self):
+        pa_schema = pa.schema([
+            ('f0', pa.int64()),
+            ('f1', pa.string()),
+        ])
 
-        cls.catalog = catalog
-        cls.df = df
+        try:
+            self.catalog.get_table('default.test_append')
+        except Exception:
+            self.catalog.create_table(
+                'default.test_append',
+                Schema(pa_schema, options={'file.format': _random_format()}),
+                True)
+            append_table = self.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(self.df)
+            commit.commit(write.prepare_commit())
+            write.close()
+            commit.close()
+
+        try:
+            self.catalog.get_table('default.test_pk')
+        except Exception:
+            self.catalog.create_table(
+                'default.test_pk',
+                Schema(pa_schema, primary_keys=['f0'],
+                       options={'bucket': '1', 'file.format': _random_format()}),
+                False)
+            pk_table = self.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(self.df)
+            commit.commit(write.prepare_commit())
+            write.close()
+            commit.close()
 
     @classmethod
     def tearDownClass(cls):
@@ -102,6 +110,7 @@ def tearDownClass(cls):
             shutil.rmtree(cls.warehouse)
 
     def testWrongFieldName(self):
+        self._init()
         table = self.catalog.get_table('default.test_append')
         predicate_builder = table.new_read_builder().new_predicate_builder()
         with self.assertRaises(ValueError) as e:
@@ -215,168 +224,196 @@ def testAllFieldTypesWithEqual(self):
         _check_filtered_result(table.new_read_builder().with_filter(predicate), df.loc[[0]])
 
     def testEqualPk(self):
+        self._init()
         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):
+        self._init()
         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):
+        self._init()
         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):
+        self._init()
         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):
+        self._init()
         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):
+        self._init()
         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):
+        self._init()
         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):
+        self._init()
         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):
+        self._init()
         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):
+        self._init()
         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):
+        self._init()
         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):
+        self._init()
         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):
+        self._init()
         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):
+        self._init()
         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):
+        self._init()
         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):
+        self._init()
         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):
+        self._init()
         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):
+        self._init()
         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):
+        self._init()
         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):
+        self._init()
         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):
+        self._init()
         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):
+        self._init()
         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):
+        self._init()
         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):
+        self._init()
         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):
+        self._init()
         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):
+        self._init()
         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):
+        self._init()
         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):
+        self._init()
         table = self.catalog.get_table('default.test_append')
         predicate_builder = table.new_read_builder().new_predicate_builder()
         predicate1 = predicate_builder.greater_than('f0', 1)
@@ -385,6 +422,7 @@ def testAndPredicates(self):
         _check_filtered_result(table.new_read_builder().with_filter(predicate), self.df.loc[[1]])
 
     def testOrPredicates(self):
+        self._init()
         table = self.catalog.get_table('default.test_append')
         predicate_builder = table.new_read_builder().new_predicate_builder()
         predicate1 = predicate_builder.greater_than('f0', 3)