[jira] [Created] (ARROW-15438) [Python] Flaky test test_write_dataset_max_open_files

["David Li (Jira)" <ji...@apache.org>]

David Li created ARROW-15438:
--------------------------------

             Summary: [Python] Flaky test test_write_dataset_max_open_files
                 Key: ARROW-15438
                 URL: https://issues.apache.org/jira/browse/ARROW-15438
             Project: Apache Arrow
          Issue Type: Bug
          Components: Python
            Reporter: David Li


Found during 7.0.0 verification
{noformat}
pyarrow/tests/test_dataset.py::test_write_dataset_max_open_files FAILED                                            [ 30%]
>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> traceback >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>tempdir = PosixPath('/tmp/pytest-of-root/pytest-1/test_write_dataset_max_open_fi0')    def test_write_dataset_max_open_files(tempdir):
        directory = tempdir / 'ds'
        file_format = "parquet"
        partition_column_id = 1
        column_names = ['c1', 'c2']
        record_batch_1 = pa.record_batch(data=[[1, 2, 3, 4, 0, 10],
                                               ['a', 'b', 'c', 'd', 'e', 'a']],
                                         names=column_names)
        record_batch_2 = pa.record_batch(data=[[5, 6, 7, 8, 0, 1],
                                               ['a', 'b', 'c', 'd', 'e', 'c']],
                                         names=column_names)
        record_batch_3 = pa.record_batch(data=[[9, 10, 11, 12, 0, 1],
                                               ['a', 'b', 'c', 'd', 'e', 'd']],
                                         names=column_names)
        record_batch_4 = pa.record_batch(data=[[13, 14, 15, 16, 0, 1],
                                               ['a', 'b', 'c', 'd', 'e', 'b']],
                                         names=column_names)
    
        table = pa.Table.from_batches([record_batch_1, record_batch_2,
                                       record_batch_3, record_batch_4])
    
        partitioning = ds.partitioning(
            pa.schema([(column_names[partition_column_id], pa.string())]),
            flavor="hive")
    
        data_source_1 = directory / "default"
    
        ds.write_dataset(data=table, base_dir=data_source_1,
                         partitioning=partitioning, format=file_format)
    
        # Here we consider the number of unique partitions created when
        # partitioning column contains duplicate records.
        #   Returns: (number_of_files_generated, number_of_partitions)
        def _get_compare_pair(data_source, record_batch, file_format, col_id):
            num_of_files_generated = _get_num_of_files_generated(
                base_directory=data_source, file_format=file_format)
            number_of_partitions = len(pa.compute.unique(record_batch[col_id]))
            return num_of_files_generated, number_of_partitions
    
        # CASE 1: when max_open_files=default & max_open_files >= num_of_partitions
        #         In case of a writing to disk via partitioning based on a
        #         particular column (considering row labels in that column),
        #         the number of unique rows must be equal
        #         to the number of files generated
    
        num_of_files_generated, number_of_partitions \
            = _get_compare_pair(data_source_1, record_batch_1, file_format,
                                partition_column_id)
        assert num_of_files_generated == number_of_partitions
    
        # CASE 2: when max_open_files > 0 & max_open_files < num_of_partitions
        #         the number of files generated must be greater than the number of
        #         partitions
    
        data_source_2 = directory / "max_1"
    
        max_open_files = 3
    
        ds.write_dataset(data=table, base_dir=data_source_2,
                         partitioning=partitioning, format=file_format,
                         max_open_files=max_open_files)
    
        num_of_files_generated, number_of_partitions \
            = _get_compare_pair(data_source_2, record_batch_1, file_format,
                                partition_column_id)
>       assert num_of_files_generated > number_of_partitions
E       assert 5 > 5pyarrow/tests/test_dataset.py:3807: AssertionErrorpyarrow/tests/test_flight.py::test_interrupt > /tmp/arrow/apache-arrow-7.0.0/python/pyarrow/tests/test_flight.py(1937)test()
-> read_all()
(Pdb) c>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> PDB continue (IO-capturing resumed) >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
FAILED                                                                [ 35%]
>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> captured stdout >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>Program interrupted. (Use 'cont' to resume).
pyarrow.lib.ArrowCancelled: Operation cancelled. Detail: received signal 2
>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> traceback >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>    def test_interrupt():
        if threading.current_thread().ident != threading.main_thread().ident:
            pytest.skip("test only works from main Python thread")
        # Skips test if not available
        raise_signal = util.get_raise_signal()
    
        def signal_from_thread():
            time.sleep(0.5)
            raise_signal(signal.SIGINT)
    
        exc_types = (KeyboardInterrupt, pa.ArrowCancelled)
    
        def test(read_all):
            try:
                try:
                    t = threading.Thread(target=signal_from_thread)
                    with pytest.raises(exc_types) as exc_info:
                        t.start()
                        read_all()
                finally:
                    t.join()
            except KeyboardInterrupt:
                # In case KeyboardInterrupt didn't interrupt read_all
                # above, at least prevent it from stopping the test suite
                pytest.fail("KeyboardInterrupt didn't interrupt Flight read_all")
            e = exc_info.value.__context__
            assert isinstance(e, pa.ArrowCancelled) or \
                isinstance(e, KeyboardInterrupt)
    
        with CancelFlightServer() as server:
            client = FlightClient(("localhost", server.port))
    
            reader = client.do_get(flight.Ticket(b""))
>           test(reader.read_all)pyarrow/tests/test_flight.py:1952: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ read_all = <built-in method read_all of pyarrow._flight.FlightStreamReader object at 0x7fe0c37590c0>    def test(read_all):
        try:
            try:
                t = threading.Thread(target=signal_from_thread)
                with pytest.raises(exc_types) as exc_info:
                    t.start()
>                   read_all()
E                   AssertionError: assert (False or False)
E                    +  where False = isinstance(None, <class 'pyarrow.lib.ArrowCancelled'>)
E                    +    where <class 'pyarrow.lib.ArrowCancelled'> = pa.ArrowCancelled
E                    +  and   False = isinstance(None, KeyboardInterrupt)pyarrow/tests/test_flight.py:1937: AssertionError
 {noformat}



--
This message was sent by Atlassian Jira
(v8.20.1#820001)