Function create_iceberg_table_from_directory

tests/common/file_utils.py:33–78

Utility function to create an iceberg table from a directory.

(impala_client, unique_database, table_name,
      file_format, table_location="${IMPALA_HOME}/testdata/data/iceberg_test",
      warehouse_prefix=os.getenv("FILESYSTEM_PREFIX"))

def create_iceberg_table_from_directory(impala_client, unique_database, table_name,
      file_format, table_location="${IMPALA_HOME}/testdata/data/iceberg_test",
      warehouse_prefix=os.getenv("FILESYSTEM_PREFIX")):
  """Utility function to create an iceberg table from a directory."""

  if not warehouse_prefix and unique_database:
    warehouse_prefix = os.getenv("DEFAULT_FS", WAREHOUSE_PREFIX)

  # Only orc, parquet, and avro tested/supported
  assert file_format in ["orc", "parquet", "avro"]

  table_location = os.path.expandvars(table_location)
  local_dir = os.path.join(table_location, table_name)
  assert os.path.isdir(local_dir)

  # Rewrite iceberg metadata to use the warehouse prefix and use unique_database
  tmp_dir = tempfile.mktemp(table_name)
  # Need to create the temp dir so 'cp -r' will copy local dir with its original name
  # under the temp dir. rewrite_metadata() has the assumption that the parent directory
  # of the 'metadata' directory bears the name of the table.
  check_call(['mkdir', '-p', tmp_dir])
  check_call(['cp', '-r', local_dir, tmp_dir])
  local_dir = os.path.join(tmp_dir, table_name)
  rewrite_metadata(warehouse_prefix, unique_database, os.path.join(local_dir, 'metadata'))

  # Put the directory in the database's directory (not the table directory)
  hdfs_parent_dir = os.path.join(get_fs_path("/test-warehouse"), unique_database + ".db")
  hdfs_dir = os.path.join(hdfs_parent_dir, table_name)

  # Purge existing files if any
  check_call(['hdfs', 'dfs', '-rm', '-skipTrash', '-f', '-r', hdfs_dir])

  # Note: -d skips a staging copy
  check_call(['hdfs', 'dfs', '-mkdir', '-p', hdfs_parent_dir])
  check_call(['hdfs', 'dfs', '-put', '-d', local_dir, hdfs_parent_dir])

  # Create external table
  qualified_table_name = '{0}.{1}'.format(unique_database, table_name)
  impala_client.execute("""create external table {0} stored as iceberg location '{1}'
                        tblproperties('write.format.default'='{2}', 'iceberg.catalog'=
                        'hadoop.tables')""".format(qualified_table_name, hdfs_dir,
                                                   file_format))

  # Automatic clean up after drop table
  impala_client.execute("""alter table {0} set tblproperties ('external.table.purge'=
                        'True');""".format(qualified_table_name))


def create_table_from_parquet(impala_client, unique_database, table_name):