fix usage of options._allow_large_results

Author: TrevorBergeron

bigframes/core/blocks.py

@@ -545,7 +545,7 @@ def to_arrow(
         under_10gb = (
             (not allow_large_results)
             if (allow_large_results is not None)
-            else bigframes.options._allow_large_results
+            else not bigframes.options._allow_large_results
         )
         execute_result = self.session._executor.execute(
             self.expr,
@@ -645,7 +645,7 @@ def try_peek(
             under_10gb = (
                 (not allow_large_results)
                 if (allow_large_results is not None)
-                else bigframes.options._allow_large_results
+                else not bigframes.options._allow_large_results
             )
             result = self.session._executor.execute(
                 self.expr,
@@ -670,7 +670,7 @@ def to_pandas_batches(
         under_10gb = (
             (not allow_large_results)
             if (allow_large_results is not None)
-            else bigframes.options._allow_large_results
+            else not bigframes.options._allow_large_results
         )
         execute_result = self.session._executor.execute(
             self.expr,
@@ -726,13 +726,13 @@ def _materialize_local(
         under_10gb = (
             (not materialize_options.allow_large_results)
             if (materialize_options.allow_large_results is not None)
-            else bigframes.options._allow_large_results
+            else (not bigframes.options._allow_large_results)
         )
         execute_result = self.session._executor.execute(
             self.expr,
             execution_spec.ExecutionSpec(
                 promise_under_10gb=under_10gb,
-                ordered=True,
+                ordered=materialize_options.ordered,
             ),
         )
         sample_config = materialize_options.downsampling

bigframes/session/bq_caching_executor.py

@@ -256,7 +256,9 @@ def _maybe_find_existing_table(
             if spec.if_exists == "fail":
                 raise ValueError(f"Table already exists: {spec.table.__str__()}")
 
-            if table.clustering_fields != spec.cluster_cols:
+            if (len(spec.cluster_cols) != 0) and (
+                table.clustering_fields != spec.cluster_cols
+            ):
                 raise ValueError(
                     "Table clustering fields cannot be changed after the table has "
                     f"been created. Existing clustering fields: {table.clustering_fields}"