Remove s3 eventual consistency guardrails. #464

Author: igorborgest

README.md

@@ -161,8 +161,8 @@ Please [send a Pull Request](https://github.com/awslabs/aws-data-wrangler/edit/m
 
 ## Amazon SageMaker Data Wrangler?
 
-**Amazon SageMaker Data Wrangler** is a new SageMaker Studio feature that has a similar name but has nothing to do with the **AWS Data Wrangler** project.
+**Amazon SageMaker Data Wrangler** is a new SageMaker Studio feature that has a similar name but has a different purpose than the **AWS Data Wrangler** open source project.
 
 * **AWS Data Wrangler** is open source, runs anywhere, and is focused on code.
 
-* **Amazon SageMaker Data Wrangler** is specific for the SageMaker environment, it is **NOT** open source, and is focused on a visual interface.
+* **Amazon SageMaker Data Wrangler** is specific for the SageMaker Studio environment and is focused on a visual interface.

awswrangler/athena/_read.py

@@ -221,11 +221,9 @@ def _fetch_parquet_result(
     metadata_path: str = manifest_path.replace("-manifest.csv", ".metadata")
     _logger.debug("manifest_path: %s", manifest_path)
     _logger.debug("metadata_path: %s", metadata_path)
-    s3.wait_objects_exist(paths=[manifest_path], use_threads=False, boto3_session=boto3_session)
     paths: List[str] = _extract_ctas_manifest_paths(path=manifest_path, boto3_session=boto3_session)
     if not paths:
         return _empty_dataframe_response(bool(chunked), query_metadata)
-    s3.wait_objects_exist(paths=paths, use_threads=False, boto3_session=boto3_session)
     ret = s3.read_parquet(
         path=paths, use_threads=use_threads, boto3_session=boto3_session, chunked=chunked, categories=categories
     )
@@ -257,7 +255,6 @@ def _fetch_csv_result(
         chunked = _chunksize is not None
         return _empty_dataframe_response(chunked, query_metadata)
     path: str = query_metadata.output_location
-    s3.wait_objects_exist(paths=[path], use_threads=False, boto3_session=boto3_session)
     _logger.debug("Start CSV reading from %s", path)
     ret = s3.read_csv(
         path=[path],

awswrangler/athena/_utils.py

@@ -121,7 +121,6 @@ def _fetch_txt_result(
     if query_metadata.output_location is None or query_metadata.output_location.endswith(".txt") is False:
         return pd.DataFrame()
     path: str = query_metadata.output_location
-    s3.wait_objects_exist(paths=[path], use_threads=False, boto3_session=boto3_session)
     _logger.debug("Start TXT reading from %s", path)
     df = s3.read_csv(
         path=[path],

tests/test__routines.py

@@ -14,7 +14,7 @@ def test_routine_0(glue_database, glue_table, path, use_threads, concurrent_part
 
     # Round 1 - Warm up
     df = pd.DataFrame({"c0": [0, None]}, dtype="Int64")
-    paths = wr.s3.to_parquet(
+    wr.s3.to_parquet(
         df=df,
         path=path,
         dataset=True,
@@ -26,9 +26,8 @@ def test_routine_0(glue_database, glue_table, path, use_threads, concurrent_part
         columns_comments={"c0": "0"},
         use_threads=use_threads,
         concurrent_partitioning=concurrent_partitioning,
-    )["paths"]
+    )
     assert wr.catalog.get_table_number_of_versions(table=glue_table, database=glue_database) == 1
-    wr.s3.wait_objects_exist(paths=paths, use_threads=use_threads)
     df2 = wr.athena.read_sql_table(glue_table, glue_database, use_threads=use_threads)
     assert df.shape == df2.shape
     assert df.c0.sum() == df2.c0.sum()
@@ -43,7 +42,7 @@ def test_routine_0(glue_database, glue_table, path, use_threads, concurrent_part
 
     # Round 2 - Overwrite
     df = pd.DataFrame({"c1": [None, 1, None]}, dtype="Int16")
-    paths = wr.s3.to_parquet(
+    wr.s3.to_parquet(
         df=df,
         path=path,
         dataset=True,
@@ -55,9 +54,8 @@ def test_routine_0(glue_database, glue_table, path, use_threads, concurrent_part
         columns_comments={"c1": "1"},
         use_threads=use_threads,
         concurrent_partitioning=concurrent_partitioning,
-    )["paths"]
+    )
     assert wr.catalog.get_table_number_of_versions(table=glue_table, database=glue_database) == 1
-    wr.s3.wait_objects_exist(paths=paths, use_threads=use_threads)
     df2 = wr.athena.read_sql_table(glue_table, glue_database, use_threads=use_threads)
     assert df.shape == df2.shape
     assert df.c1.sum() == df2.c1.sum()
@@ -72,7 +70,7 @@ def test_routine_0(glue_database, glue_table, path, use_threads, concurrent_part
 
     # Round 3 - Append
     df = pd.DataFrame({"c1": [None, 2, None]}, dtype="Int8")
-    paths = wr.s3.to_parquet(
+    wr.s3.to_parquet(
         df=df,
         path=path,
         dataset=True,
@@ -84,9 +82,8 @@ def test_routine_0(glue_database, glue_table, path, use_threads, concurrent_part
         columns_comments={"c1": "1"},
         use_threads=use_threads,
         concurrent_partitioning=concurrent_partitioning,
-    )["paths"]
+    )
     assert wr.catalog.get_table_number_of_versions(table=glue_table, database=glue_database) == 1
-    wr.s3.wait_objects_exist(paths=paths, use_threads=use_threads)
     df2 = wr.athena.read_sql_table(glue_table, glue_database, use_threads=use_threads)
     assert len(df.columns) == len(df2.columns)
     assert len(df.index) * 2 == len(df2.index)
@@ -102,7 +99,7 @@ def test_routine_0(glue_database, glue_table, path, use_threads, concurrent_part
 
     # Round 4 - Append + New Column
     df = pd.DataFrame({"c2": ["a", None, "b"], "c1": [None, None, None]})
-    paths = wr.s3.to_parquet(
+    wr.s3.to_parquet(
         df=df,
         path=path,
         dataset=True,
@@ -114,9 +111,8 @@ def test_routine_0(glue_database, glue_table, path, use_threads, concurrent_part
         columns_comments={"c1": "1", "c2": "2"},
         use_threads=use_threads,
         concurrent_partitioning=concurrent_partitioning,
-    )["paths"]
+    )
     assert wr.catalog.get_table_number_of_versions(table=glue_table, database=glue_database) == 1
-    wr.s3.wait_objects_exist(paths=paths, use_threads=use_threads)
     df2 = wr.athena.read_sql_table(glue_table, glue_database, use_threads=use_threads)
     assert len(df2.columns) == 2
     assert len(df2.index) == 9
@@ -133,7 +129,7 @@ def test_routine_0(glue_database, glue_table, path, use_threads, concurrent_part
 
     # Round 5 - Append + New Column + Wrong Types
     df = pd.DataFrame({"c2": [1], "c3": [True], "c1": ["1"]})
-    paths = wr.s3.to_parquet(
+    wr.s3.to_parquet(
         df=df,
         path=path,
         dataset=True,
@@ -145,9 +141,8 @@ def test_routine_0(glue_database, glue_table, path, use_threads, concurrent_part
         columns_comments={"c1": "1!", "c2": "2!", "c3": "3"},
         use_threads=use_threads,
         concurrent_partitioning=concurrent_partitioning,
-    )["paths"]
+    )
     assert wr.catalog.get_table_number_of_versions(table=glue_table, database=glue_database) == 1
-    wr.s3.wait_objects_exist(paths=paths, use_threads=use_threads)
     df2 = wr.athena.read_sql_table(glue_table, glue_database, use_threads=use_threads)
     assert len(df2.columns) == 3
     assert len(df2.index) == 10
@@ -165,7 +160,7 @@ def test_routine_0(glue_database, glue_table, path, use_threads, concurrent_part
 
     # Round 6 - Overwrite Partitioned
     df = pd.DataFrame({"c0": ["foo", None], "c1": [0, 1]})
-    paths = wr.s3.to_parquet(
+    wr.s3.to_parquet(
         df=df,
         path=path,
         dataset=True,
@@ -178,9 +173,8 @@ def test_routine_0(glue_database, glue_table, path, use_threads, concurrent_part
         columns_comments={"c0": "zero", "c1": "one"},
         use_threads=use_threads,
         concurrent_partitioning=concurrent_partitioning,
-    )["paths"]
+    )
     assert wr.catalog.get_table_number_of_versions(table=glue_table, database=glue_database) == 1
-    wr.s3.wait_objects_exist(paths=paths, use_threads=use_threads)
     df2 = wr.athena.read_sql_table(glue_table, glue_database, use_threads=use_threads)
     assert df.shape == df2.shape
     assert df.c1.sum() == df2.c1.sum()
@@ -196,7 +190,7 @@ def test_routine_0(glue_database, glue_table, path, use_threads, concurrent_part
 
     # Round 7 - Overwrite Partitions
     df = pd.DataFrame({"c0": [None, None], "c1": [0, 2]})
-    paths = wr.s3.to_parquet(
+    wr.s3.to_parquet(
         df=df,
         path=path,
         dataset=True,
@@ -209,9 +203,8 @@ def test_routine_0(glue_database, glue_table, path, use_threads, concurrent_part
         columns_comments={"c0": "zero", "c1": "one"},
         concurrent_partitioning=concurrent_partitioning,
         use_threads=use_threads,
-    )["paths"]
+    )
     assert wr.catalog.get_table_number_of_versions(table=glue_table, database=glue_database) == 1
-    wr.s3.wait_objects_exist(paths=paths, use_threads=use_threads)
     df2 = wr.athena.read_sql_table(glue_table, glue_database, use_threads=use_threads)
     assert len(df2.columns) == 2
     assert len(df2.index) == 3
@@ -228,7 +221,7 @@ def test_routine_0(glue_database, glue_table, path, use_threads, concurrent_part
 
     # Round 8 - Overwrite Partitions + New Column + Wrong Type
     df = pd.DataFrame({"c0": [1, 2], "c1": ["1", "3"], "c2": [True, False]})
-    paths = wr.s3.to_parquet(
+    wr.s3.to_parquet(
         df=df,
         path=path,
         dataset=True,
@@ -241,9 +234,8 @@ def test_routine_0(glue_database, glue_table, path, use_threads, concurrent_part
         columns_comments={"c0": "zero", "c1": "one", "c2": "two"},
         use_threads=use_threads,
         concurrent_partitioning=concurrent_partitioning,
-    )["paths"]
+    )
     assert wr.catalog.get_table_number_of_versions(table=glue_table, database=glue_database) == 1
-    wr.s3.wait_objects_exist(paths=paths, use_threads=use_threads)
     df2 = wr.athena.read_sql_table(glue_table, glue_database, use_threads=use_threads)
     assert len(df2.columns) == 3
     assert len(df2.index) == 4
@@ -264,8 +256,7 @@ def test_routine_1(glue_database, glue_table, path):
 
     # Round 1 - Warm up
     df = pd.DataFrame({"c0": [0, None]}, dtype="Int64")
-    paths = wr.s3.to_parquet(df=df, path=path, dataset=True, mode="overwrite")["paths"]
-    wr.s3.wait_objects_exist(paths=paths)
+    wr.s3.to_parquet(df=df, path=path, dataset=True, mode="overwrite")
     wr.s3.store_parquet_metadata(
         path=path,
         dataset=True,
@@ -291,8 +282,7 @@ def test_routine_1(glue_database, glue_table, path):
 
     # Round 2 - Overwrite
     df = pd.DataFrame({"c1": [None, 1, None]}, dtype="Int16")
-    paths = wr.s3.to_parquet(df=df, path=path, dataset=True, mode="overwrite")["paths"]
-    wr.s3.wait_objects_exist(paths=paths)
+    wr.s3.to_parquet(df=df, path=path, dataset=True, mode="overwrite")
     wr.s3.store_parquet_metadata(
         path=path,
         dataset=True,
@@ -318,8 +308,7 @@ def test_routine_1(glue_database, glue_table, path):
 
     # Round 3 - Append
     df = pd.DataFrame({"c1": [None, 2, None]}, dtype="Int16")
-    paths = wr.s3.to_parquet(df=df, path=path, dataset=True, mode="append")["paths"]
-    wr.s3.wait_objects_exist(paths=paths)
+    wr.s3.to_parquet(df=df, path=path, dataset=True, mode="append")
     wr.s3.store_parquet_metadata(
         path=path,
         dataset=True,
@@ -347,8 +336,7 @@ def test_routine_1(glue_database, glue_table, path):
     # Round 4 - Append + New Column
     df = pd.DataFrame({"c2": ["a", None, "b"], "c1": [None, 1, None]})
     df["c1"] = df["c1"].astype("Int16")
-    paths = wr.s3.to_parquet(df=df, path=path, dataset=True, mode="append")["paths"]
-    wr.s3.wait_objects_exist(paths=paths)
+    wr.s3.to_parquet(df=df, path=path, dataset=True, mode="append")
     wr.s3.store_parquet_metadata(
         path=path,
         dataset=True,
@@ -376,8 +364,7 @@ def test_routine_1(glue_database, glue_table, path):
 
     # Round 5 - Overwrite Partitioned
     df = pd.DataFrame({"c0": ["foo", None], "c1": [0, 1]})
-    paths = wr.s3.to_parquet(df=df, path=path, dataset=True, mode="overwrite", partition_cols=["c1"])["paths"]
-    wr.s3.wait_objects_exist(paths=paths)
+    wr.s3.to_parquet(df=df, path=path, dataset=True, mode="overwrite", partition_cols=["c1"])
     wr.s3.store_parquet_metadata(
         path=path,
         dataset=True,
@@ -404,10 +391,7 @@ def test_routine_1(glue_database, glue_table, path):
 
     # Round 6 - Overwrite Partitions
     df = pd.DataFrame({"c0": [None, "boo"], "c1": [0, 2]})
-    paths = wr.s3.to_parquet(df=df, path=path, dataset=True, mode="overwrite_partitions", partition_cols=["c1"])[
-        "paths"
-    ]
-    wr.s3.wait_objects_exist(paths=paths)
+    wr.s3.to_parquet(df=df, path=path, dataset=True, mode="overwrite_partitions", partition_cols=["c1"])
     wr.s3.store_parquet_metadata(
         path=path,
         dataset=True,
@@ -435,10 +419,7 @@ def test_routine_1(glue_database, glue_table, path):
 
     # Round 7 - Overwrite Partitions + New Column
     df = pd.DataFrame({"c0": ["bar", None], "c1": [1, 3], "c2": [True, False]})
-    paths = wr.s3.to_parquet(df=df, path=path, dataset=True, mode="overwrite_partitions", partition_cols=["c1"])[
-        "paths"
-    ]
-    wr.s3.wait_objects_exist(paths=paths)
+    wr.s3.to_parquet(df=df, path=path, dataset=True, mode="overwrite_partitions", partition_cols=["c1"])
     wr.s3.store_parquet_metadata(
         path=path,
         dataset=True,