test_integration.py 文件源码

def test_load_empty_table_arrow(self, con, empty_table):
        pd = pytest.importorskip("pandas")
        pa = pytest.importorskip("pyarrow")
        skip_if_no_arrow_loader(con)

        data = [(1, 1.1, 'a'),
                (2, 2.2, '2'),
                (3, 3.3, '3')]

        df = pd.DataFrame(data, columns=list('abc')).astype({
            'a': 'int32',
            'b': 'float32'
        })

        table = pa.Table.from_pandas(df, preserve_index=False)
        con.load_table(empty_table, table, method='arrow')
        result = sorted(con.execute("select * from {}".format(empty_table)))
        self.check_empty_insert(result, data)