path1 = tmp_path / "dir1"

path2 = tmp_path / "dir2"

path1 = str(path1) if path_to_str else path1

path2 = str(path2) if path_to_str else path2

runtime = RuntimeEnvBuilder().with_disk_manager_specified(path1, path2)

config = SessionConfig().with_default_catalog_and_schema("foo", "bar")

ctx = SessionContext(config, runtime)

assert ctx is not None

db = ctx.catalog("foo").schema("bar")

path = str(tmp_path) if path_to_str else tmp_path

runtime = RuntimeEnvBuilder().with_temp_file_path(path)

config = SessionConfig().with_default_catalog_and_schema("foo", "bar")

ctx = SessionContext(config, runtime)

assert ctx is not None

db = ctx.catalog("foo").schema("bar")

runtime = RuntimeEnvBuilder().with_disk_manager_os().with_fair_spill_pool(10000000)

config = (

SessionConfig()

.with_create_default_catalog_and_schema(enabled=True)

.with_default_catalog_and_schema("foo", "bar")

.with_target_partitions(1)

.with_information_schema(enabled=True)

.with_repartition_joins(enabled=False)

.with_repartition_aggregations(enabled=False)

.with_repartition_windows(enabled=False)

.with_parquet_pruning(enabled=False)

)

ctx = SessionContext(config, runtime)

# verify that at least some of the arguments worked

ctx.catalog("foo").schema("bar")

with pytest.raises(KeyError):

# create a RecordBatch and register it as memtable

batch = pa.RecordBatch.from_arrays(

[pa.array([1, 2, 3]), pa.array([4, 5, 6])],

names=["a", "b"],

)

ctx.register_record_batches("t", [[batch]])

assert ctx.catalog().schema().names() == {"t"}

result = ctx.sql("SELECT a+b, a-b FROM t").collect()

assert result[0].column(0) == pa.array([5, 7, 9])

# create a RecordBatch and register it as memtable

batch = pa.RecordBatch.from_arrays(

[pa.array([1, 2, 3]), pa.array([4, 5, 6])],

names=["a", "b"],

)

df = ctx.create_dataframe([[batch]])

tables = list(ctx.catalog().schema().names())

assert df

assert len(tables) == 1

assert len(tables[0]) == 33

assert tables[0].startswith("c")

# ensure that the rest of the table name contains

# only hexadecimal numbers

for c in tables[0][1:]:

# create a RecordBatch and register it as memtable

batch = pa.RecordBatch.from_arrays(

[pa.array([1, 2, 3]), pa.array([4, 5, 6])],

names=["a", "b"],

)

df = ctx.create_dataframe([[batch]], name="tbl")

tables = list(ctx.catalog().schema().names())

assert df

assert len(tables) == 1

# create a PyArrow table

data = {"a": [1, 2, 3], "b": [4, 5, 6]}

table = pa.Table.from_pydict(data)

# convert to DataFrame

df = ctx.from_arrow(table)

tables = list(ctx.catalog().schema().names())

assert df

assert len(tables) == 1

assert isinstance(df, DataFrame)

assert set(df.schema().names) == {"a", "b"}

schema = pa.schema([("a", pa.int64()), ("b", pa.int64())])

for _i in range(num_batches):

yield pa.RecordBatch.from_arrays(

[pa.array([1, 2, 3]), pa.array([4, 5, 6])], schema=schema

df = ctx.from_arrow(source)

assert df

assert isinstance(df, DataFrame)

assert df.schema().names == ["a", "b"]

# create a PyArrow table

data = {"a": [1, 2, 3], "b": [4, 5, 6]}

table = pa.Table.from_pydict(data)

# convert to DataFrame with optional name

df = ctx.from_arrow(table, name="tbl")

tables = list(ctx.catalog().schema().names())

assert df

data = {"a": [], "b": []}

schema = pa.schema([("a", pa.int32()), ("b", pa.string())])

table = pa.Table.from_pydict(data, schema=schema)

# convert to DataFrame

df = ctx.from_arrow(table)

tables = list(ctx.catalog().schema().names())

assert df

assert len(tables) == 1

assert isinstance(df, DataFrame)

assert set(df.schema().names) == {"a", "b"}

data = {"a": [], "b": []}

table = pa.Table.from_pydict(data)

# convert to DataFrame

df = ctx.from_arrow(table)

tables = list(ctx.catalog().schema().names())

assert df

assert len(tables) == 1

assert isinstance(df, DataFrame)

assert set(df.schema().names) == {"a", "b"}

# create a dataframe from Python list

data = [

{"a": 1, "b": 4},

{"a": 2, "b": 5},

{"a": 3, "b": 6},

]

df = ctx.from_pylist(data)

tables = list(ctx.catalog().schema().names())

assert df

assert len(tables) == 1

assert isinstance(df, DataFrame)

assert set(df.schema().names) == {"a", "b"}

# create a dataframe from Python dictionary

data = {"a": [1, 2, 3], "b": [4, 5, 6]}

df = ctx.from_pydict(data)

tables = list(ctx.catalog().schema().names())

assert df

assert len(tables) == 1

assert isinstance(df, DataFrame)

assert set(df.schema().names) == {"a", "b"}

# create a dataframe from pandas dataframe

pd = pytest.importorskip("pandas")

data = {"a": [1, 2, 3], "b": [4, 5, 6]}

pandas_df = pd.DataFrame(data)

df = ctx.from_pandas(pandas_df)

tables = list(ctx.catalog().schema().names())

assert df

assert len(tables) == 1

assert isinstance(df, DataFrame)

assert set(df.schema().names) == {"a", "b"}

# create a dataframe from Polars dataframe

pd = pytest.importorskip("polars")

data = {"a": [1, 2, 3], "b": [4, 5, 6]}

polars_df = pd.DataFrame(data)

df = ctx.from_polars(polars_df)

tables = list(ctx.catalog().schema().names())

assert df

assert len(tables) == 1

assert isinstance(df, DataFrame)

assert set(df.schema().names) == {"a", "b"}

default = ctx.catalog()

public = default.schema("public")

assert public.names() == {"csv", "csv1", "csv2"}

table = public.table("csv")

ctx.register_table("csv3", table)

default = ctx.catalog()

public = default.schema("public")

assert public.names() == {"csv", "csv1", "csv2"}

table = public.table("csv")

table_df = ctx.read_table(table)

batch = pa.RecordBatch.from_arrays(

[pa.array([1, 2, 3]), pa.array([4, 5, 6])],

names=["a", "b"],

)

dataset = ds.dataset([batch])

result = ctx.read_table(dataset).collect()

assert result[0].column(0) == pa.array([1, 2, 3])

default = ctx.catalog()

public = default.schema("public")

assert public.names() == {"csv", "csv1", "csv2"}

ctx.deregister_table("csv")

df = ctx.from_pydict({"a": [1, 2]})

ctx.register_table("df_tbl", df)

result = ctx.sql("SELECT * FROM df_tbl").collect()

df = ctx.from_pydict({"a": [1, 2]})

table = df.into_view(temporary=temporary)

assert isinstance(table, Table)

if temporary:

assert table.kind == "temporary"

else:

assert table.kind == "view"

ctx.register_table("view_tbl", table)

result = ctx.sql("SELECT * FROM view_tbl").collect()

df = ctx.from_pydict({"a": [1, 2]})

table = Table(df)

assert isinstance(table, Table)

ctx.register_table("from_dataframe_tbl", table)

result = ctx.sql("SELECT * FROM from_dataframe_tbl").collect()

df = ctx.from_pydict({"a": [1, 2]})

table = Table(df.df)

assert isinstance(table, Table)

ctx.register_table("from_internal_dataframe_tbl", table)

result = ctx.sql("SELECT * FROM from_internal_dataframe_tbl").collect()

# create a RecordBatch and register it as a pyarrow.dataset.Dataset

batch = pa.RecordBatch.from_arrays(

[pa.array([1, 2, 3]), pa.array([4, 5, 6])],

names=["a", "b"],

)

dataset = ds.dataset([batch])

ctx.register_dataset("t", dataset)

assert ctx.catalog().schema().names() == {"t"}

result = ctx.sql("SELECT a+b, a-b FROM t").collect()

assert result[0].column(0) == pa.array([5, 7, 9])

# create a RecordBatch and register it as a pyarrow.dataset.Dataset

batch = pa.RecordBatch.from_arrays(

[pa.array([1, 2, 3]), pa.array([4, 5, 6])],

names=["a", "b"],

)

dataset = ds.dataset([batch])

ctx.register_dataset("t", dataset)

assert ctx.catalog().schema().names() == {"t"}

df = ctx.sql("SELECT a+b, a-b FROM t WHERE a BETWEEN 2 and 3 AND b > 5")

# Make sure the filter was pushed down in Physical Plan

df.explain()

captured = capfd.readouterr()

assert "filter_expr=(((a >= 2) and (a <= 3)) and (b > 5))" in captured.out

result = df.collect()

assert result[0].column(0) == pa.array([9])

# `datafusion-python` issue: https://github.com/apache/datafusion-python/issues/800

batch = pa.RecordBatch.from_arrays(

[pa.array([1, 2, 3]), pa.array([4, 5, 6])],

names=["a", "b"],

)

dataset = ds.dataset([batch])

ctx.register_dataset("t", dataset)

# Testing the dataframe API

df = ctx.table("t")

assert df.count() == 3

# Testing the SQL API

count = ctx.sql("SELECT COUNT(*) FROM t")

count = count.collect()

"""Ensure that pyarrow filter gets pushed down for `IsNull`"""

# create a RecordBatch and register it as a pyarrow.dataset.Dataset

batch = pa.RecordBatch.from_arrays(

[pa.array([1, 2, 3]), pa.array([4, 5, 6]), pa.array([7, None, 9])],

names=["a", "b", "c"],

)

dataset = ds.dataset([batch])

ctx.register_dataset("t", dataset)

# Make sure the filter was pushed down in Physical Plan

df = ctx.sql("SELECT a FROM t WHERE c is NULL")

df.explain()

captured = capfd.readouterr()

assert "filter_expr=is_null(c, {nan_is_null=false})" in captured.out

result = df.collect()

"""Ensure that pyarrow filter gets pushed down for timestamp"""

# Ref: https://github.com/apache/datafusion-python/issues/703

# create pyarrow dataset with no actual files

col_type = pa.timestamp("ns", "+00:00")

nyd_2000 = pa.scalar(dt.datetime(2000, 1, 1, tzinfo=dt.timezone.utc), col_type)

pa_dataset_fs = pa.fs.SubTreeFileSystem(str(tmpdir), pa.fs.LocalFileSystem())

pa_dataset_format = pa.dataset.ParquetFileFormat()

pa_dataset_partition = pa.dataset.field("a") <= nyd_2000

fragments = [

# NOTE: we never actually make this file.

# Working predicate pushdown means it never gets accessed

pa_dataset_format.make_fragment(

"1.parquet",

filesystem=pa_dataset_fs,

partition_expression=pa_dataset_partition,

)

]

pa_dataset = pa.dataset.FileSystemDataset(

fragments,

pa.schema([pa.field("a", col_type)]),

pa_dataset_format,

pa_dataset_fs,

)

ctx.register_dataset("t", pa_dataset)

# the partition for our only fragment is for a < 2000-01-01.

# so querying for a > 2024-01-01 should not touch any files

df = ctx.sql("SELECT * FROM t WHERE a > '2024-01-01T00:00:00+00:00'")

# create Arrow StructArrays to test nested data types

data = pa.StructArray.from_arrays(

[pa.array([1, 2, 3]), pa.array([4, 5, 6])],

names=["a", "b"],

)

batch = pa.RecordBatch.from_arrays(

[data],

names=["nested_data"],

)

dataset = ds.dataset([batch])

ctx.register_dataset("t", dataset)

assert ctx.catalog().schema().names() == {"t"}

df = ctx.table("t")

# This filter will not be pushed down to DatasetExec since it

# isn't supported

df = df.filter(column("nested_data")["b"] > literal(5)).select(

column("nested_data")["a"] + column("nested_data")["b"],

column("nested_data")["a"] - column("nested_data")["b"],

)

result = df.collect()

assert result[0].column(0) == pa.array([9])

batch = pa.RecordBatch.from_arrays(

[pa.array([1, 2, 3]), pa.array([4, 5, 6])],

names=["a", "b"],

)

dataset = ds.dataset([batch])

ctx.register_dataset("t", dataset)

from uuid import uuid4

with pytest.raises(KeyError):

path = pathlib.Path(__file__).parent.resolve()

# Default

test_data_path = path / "data_test_context" / "data.json"

df = ctx.read_json(test_data_path)

result = df.collect()

assert result[0].column(0) == pa.array(["a", "b", "c"])

assert result[0].column(1) == pa.array([1, 2, 3])

# Schema

schema = pa.schema(

[

pa.field("A", pa.string(), nullable=True),

]

)

df = ctx.read_json(test_data_path, schema=schema)

result = df.collect()

assert result[0].column(0) == pa.array(["a", "b", "c"])

assert result[0].schema == schema

# File extension

test_data_path = path / "data_test_context" / "data.json"

df = ctx.read_json(test_data_path, file_extension=".json")

result = df.collect()

assert result[0].column(0) == pa.array(["a", "b", "c"])

path = pathlib.Path(__file__).parent.resolve()

test_data_path = path / "data_test_context" / "data.json"

# File compression type

gzip_path = tmp_path / "data.json.gz"

with (

pathlib.Path.open(test_data_path, "rb") as csv_file,

gzip.open(gzip_path, "wb") as gzipped_file,

):

gzipped_file.writelines(csv_file)

df = ctx.read_json(gzip_path, file_extension=".gz", file_compression_type="gz")

result = df.collect()

assert result[0].column(0) == pa.array(["a", "b", "c"])

csv_df = ctx.read_csv(path=["testing/data/csv/aggregate_test_100.csv"])

expected = csv_df.count() * 2

double_csv_df = ctx.read_csv(

path=[

"testing/data/csv/aggregate_test_100.csv",

"testing/data/csv/aggregate_test_100.csv",

]

)

actual = double_csv_df.count()

double_csv_df.select(column("c1")).show()

test_data_path = pathlib.Path("testing/data/csv/aggregate_test_100.csv")

expected = ctx.read_csv(test_data_path).collect()

# File compression type

gzip_path = tmp_path / "aggregate_test_100.csv.gz"

with (

pathlib.Path.open(test_data_path, "rb") as csv_file,

gzip.open(gzip_path, "wb") as gzipped_file,

):

gzipped_file.writelines(csv_file)

csv_df = ctx.read_csv(gzip_path, file_extension=".gz", file_compression_type="gz")

assert csv_df.collect() == expected

csv_df = ctx.read_csv(

gzip_path,

options=CsvReadOptions(file_extension=".gz", file_compression_type="gz"),

)

parquet_df = ctx.read_parquet(path="parquet/data/alltypes_plain.parquet")

parquet_df.show()

assert parquet_df is not None

path = pathlib.Path.cwd() / "parquet/data/alltypes_plain.parquet"

parquet_df = ctx.read_parquet(path=path)

avro_df = ctx.read_avro(path="testing/data/avro/alltypes_plain.avro")

avro_df.show()

assert avro_df is not None

path = pathlib.Path.cwd() / "testing/data/avro/alltypes_plain.avro"

avro_df = ctx.read_avro(path=path)

sql = "CREATE TABLE IF NOT EXISTS valuetable AS VALUES(1,'HELLO'),(12,'DATAFUSION')"

ctx.sql(sql)

options = SQLOptions().with_allow_ddl(allow=False)

with pytest.raises(Exception, match="DDL"):

table_name = "t"

batch = pa.RecordBatch.from_arrays(

[pa.array([1, 2, 3]), pa.array([4, 5, 6])],

names=["a", "b"],

)

dataset = ds.dataset([batch])

ctx.register_dataset(table_name, dataset)

sql = f'INSERT INTO "{table_name}" VALUES (1, 2), (2, 3);'

ctx.sql(sql)

options = SQLOptions().with_allow_dml(allow=False)

with pytest.raises(Exception, match="DML"):

sql = "SET time zone = 1;"

ctx.sql(sql)

options = SQLOptions().with_allow_statements(allow=False)

with pytest.raises(Exception, match="SetVariable"):

return pa.RecordBatch.from_arrays(

[pa.array([4, 5, 6])],

names=["a"],

ctx = SessionContext.global_ctx()

df = ctx.create_dataframe([[batch]])

result = df.collect()[0].column(0)

"""Test CsvReadOptions builder pattern."""

from datafusion import CsvReadOptions

options = (

CsvReadOptions()

.with_has_header(False) # noqa: FBT003

.with_delimiter("|")

.with_quote("'")

.with_schema_infer_max_records(2000)

.with_truncated_rows(True) # noqa: FBT003

.with_newlines_in_values(True) # noqa: FBT003

.with_file_extension(".tsv")

)

assert options.has_header is False

assert options.delimiter == "|"

assert options.quote == "'"

assert options.schema_infer_max_records == 2000

assert options.truncated_rows is True

assert options.newlines_in_values is True

tmp_path: pathlib.Path,

csv_content: str,

options: CsvReadOptions,

expected: pa.RecordBatch,

as_read: bool,

global_ctx: bool,

) -> None:

from datafusion import SessionContext

# Create a test CSV file

group_dir = tmp_path / "group=a"

group_dir.mkdir(exist_ok=True)

csv_path = group_dir / "test.csv"

csv_path.write_text(csv_content, newline="\n")

ctx = SessionContext()

if as_read:

if global_ctx:

from datafusion.io import read_csv

df = read_csv(str(tmp_path), options=options)

else:

df = ctx.read_csv(str(tmp_path), options=options)

else:

ctx.register_csv("test_table", str(tmp_path), options=options)

df = ctx.sql("SELECT * FROM test_table")

df.show()

# Verify the data

result = df.collect()

assert len(result) == 1

"""Test reading CSV with CsvReadOptions."""

csv_content = "Alice;30;|New York; NY|\nBob;25\n#Charlie;35;Paris\nPhil;75;Detroit' MI\nKarin;50;|Stockholm\nSweden|" # noqa: E501

# Some of the read options are difficult to test in combination

# such as schema and schema_infer_max_records so run multiple tests

# file_sort_order doesn't impact reading, but included here to ensure

# all options parse correctly

options = CsvReadOptions(

has_header=False,

delimiter=";",

quote="|",

terminator="\n",

escape="\\",

comment="#",

newlines_in_values=True,

schema_infer_max_records=1,

null_regex="[pP]+aris",

truncated_rows=True,

file_sort_order=[[column("column_1").sort(), column("column_2")], ["column_3"]],

)

expected = pa.RecordBatch.from_arrays(

[

pa.array(["Alice", "Bob", "Phil", "Karin"]),

pa.array([30, 25, 75, 50]),

pa.array(["New York; NY", None, "Detroit' MI", "Stockholm\nSweden"]),

],

names=["column_1", "column_2", "column_3"],

)

read_csv_with_options_inner(

tmp_path, csv_content, options, expected, as_read, global_ctx

)

schema = pa.schema(

[

pa.field("name", pa.string(), nullable=False),

pa.field("age", pa.float32(), nullable=False),

pa.field("location", pa.string(), nullable=True),

]

)

options.with_schema(schema)

expected = pa.RecordBatch.from_arrays(

[

pa.array(["Alice", "Bob", "Phil", "Karin"]),

pa.array([30.0, 25.0, 75.0, 50.0]),

pa.array(["New York; NY", None, "Detroit' MI", "Stockholm\nSweden"]),

],

schema=schema,

)

read_csv_with_options_inner(

tmp_path, csv_content, options, expected, as_read, global_ctx

)

csv_content = "name,age\nAlice,30\nBob,25\nCharlie,35\nDiego,40\nEmily,15"

expected = pa.RecordBatch.from_arrays(

[

pa.array(["Alice", "Bob", "Charlie", "Diego", "Emily"]),

pa.array([30, 25, 35, 40, 15]),

pa.array(["a", "a", "a", "a", "a"]),

],

schema=pa.schema(

[

pa.field("name", pa.string(), nullable=True),

pa.field("age", pa.int64(), nullable=True),

pa.field("group", pa.string(), nullable=False),

]

),

)

options = CsvReadOptions(

table_partition_cols=[("group", pa.string())],

)

read_csv_with_options_inner(

tmp_path, csv_content, options, expected, as_read, global_ctx