Snow 2220426 streaming output (#2589)

* SNOW-2220426: streaming output

* SNOW-2220426: streaming output release notes

* SNOW-2220426: streaming output fix new lines handling

* SNOW-2220426: streaming output fix windows paths in test

* SNOW-2220426: streaming output drop deprecated function

Author: sfc-gh-mraba

RELEASE-NOTES.md

@@ -27,6 +27,7 @@
 * Bumped to `typer=0.17.3`. Improved displaying help messages.
 * Fixed using `ctx.var` in `snow sql` with Jinja templating.
 * Fixed issues when pasting content with trailing new lines.
+* Improved output handling  with streaming
 
 
 # v3.11.0

src/snowflake/cli/_app/printing.py

@@ -22,7 +22,7 @@
 from json import JSONEncoder
 from pathlib import Path
 from textwrap import indent
-from typing import TextIO
+from typing import Any, Dict, TextIO
 
 from rich import box, get_console
 from rich import print as rich_print
@@ -61,13 +61,114 @@ def default(self, o):
             return list(o.result)
         if isinstance(o, (date, datetime, time)):
             return o.isoformat()
-        if isinstance(o, (Path, Decimal)):
+        if isinstance(o, Path):
+            return o.as_posix()
+        if isinstance(o, Decimal):
             return str(o)
         if isinstance(o, bytearray):
             return o.hex()
         return super().default(o)
 
 
+class StreamingJSONEncoder(JSONEncoder):
+    """Streaming JSON encoder that doesn't materialize generators into lists"""
+
+    def default(self, o):
+        if isinstance(o, str):
+            return sanitize_for_terminal(o)
+        if isinstance(o, (ObjectResult, MessageResult)):
+            return o.result
+        if isinstance(o, (CollectionResult, MultipleResults)):
+            raise TypeError(
+                f"CollectionResult should be handled by streaming functions, not encoder"
+            )
+        if isinstance(o, (date, datetime, time)):
+            return o.isoformat()
+        if isinstance(o, Path):
+            return o.as_posix()
+        if isinstance(o, Decimal):
+            return str(o)
+        if isinstance(o, bytearray):
+            return o.hex()
+        return super().default(o)
+
+
+def _print_json_item_with_array_indentation(item: Any, indent: int):
+    """Print a JSON item with proper indentation for array context"""
+    if indent:
+        indented_output = json.dumps(item, cls=StreamingJSONEncoder, indent=indent)
+        indented_lines = indented_output.split("\n")
+        for i, line in enumerate(indented_lines):
+            if i == 0:
+                print(" " * indent + line, end="")
+            else:
+                print("\n" + " " * indent + line, end="")
+    else:
+        json.dump(item, sys.stdout, cls=StreamingJSONEncoder, separators=(",", ":"))
+
+
+def _stream_collection_as_json(result: CollectionResult, indent: int = 4):
+    """Stream a CollectionResult as a JSON array without loading all data into memory"""
+    items = iter(result.result)
+    try:
+        first_item = next(items)
+    except StopIteration:
+        print("[]", end="")
+        return
+
+    print("[")
+
+    _print_json_item_with_array_indentation(first_item, indent)
+
+    for item in items:
+        print(",")
+        _print_json_item_with_array_indentation(item, indent)
+
+    print("\n]", end="")
+
+
+def _stream_collection_as_csv(result: CollectionResult):
+    """Stream a CollectionResult as CSV without loading all data into memory"""
+    items = iter(result.result)
+    try:
+        first_item = next(items)
+    except StopIteration:
+        return
+
+    fieldnames = list(first_item.keys())
+    if not isinstance(first_item, dict):
+        raise TypeError("CSV output requires dictionary items")
+
+    writer = csv.DictWriter(sys.stdout, fieldnames=fieldnames, lineterminator="\n")
+    writer.writeheader()
+    _write_csv_row(writer, first_item)
+
+    for item in items:
+        _write_csv_row(writer, item)
+
+
+def _write_csv_row(writer: csv.DictWriter, row_data: Dict[str, Any]):
+    """Write a single CSV row, handling special data types"""
+    processed_row = {}
+    for key, value in row_data.items():
+        if isinstance(value, str):
+            processed_row[key] = sanitize_for_terminal(value)
+        elif isinstance(value, (date, datetime, time)):
+            processed_row[key] = value.isoformat()
+        elif isinstance(value, Path):
+            processed_row[key] = value.as_posix()
+        elif isinstance(value, Decimal):
+            processed_row[key] = str(value)
+        elif isinstance(value, bytearray):
+            processed_row[key] = value.hex()
+        elif value is None:
+            processed_row[key] = ""
+        else:
+            processed_row[key] = str(value)
+
+    writer.writerow(processed_row)
+
+
 def _get_format_type() -> OutputFormat:
     output_format = get_cli_context().output_format
     if output_format:
@@ -110,12 +211,13 @@ def is_structured_format(output_format):
 def print_structured(
     result: CommandResult, output_format: OutputFormat = OutputFormat.JSON
 ):
-    """Handles outputs like json, yml and other structured and parsable formats."""
+    """Handles outputs like json, csv and other structured and parsable formats with streaming."""
     printed_end_line = False
+
     if isinstance(result, MultipleResults):
         if output_format == OutputFormat.CSV:
             for command_result in result.result:
-                _print_csv_result(command_result)
+                _print_csv_result_streaming(command_result)
                 print(flush=True)
             printed_end_line = True
         else:
@@ -125,35 +227,67 @@ def print_structured(
         # instead of joining all the values into a JSON array or CSV entry set
         for r in result.result:
             if output_format == OutputFormat.CSV:
-                _print_csv_result(r.result)
+                _print_csv_result_streaming(r)
             else:
-                json.dump(r, sys.stdout, cls=CustomJSONEncoder)
+                json.dump(r, sys.stdout, cls=StreamingJSONEncoder)
             print(flush=True)
             printed_end_line = True
     else:
         if output_format == OutputFormat.CSV:
-            _print_csv_result(result)
+            _print_csv_result_streaming(result)
             printed_end_line = True
         else:
-            json.dump(result, sys.stdout, cls=CustomJSONEncoder, indent=4)
+            _print_json_result_streaming(result)
+
     # Adds empty line at the end
     if not printed_end_line:
         print(flush=True)
 
 
-def _print_csv_result(result: CommandResult):
-    data = json.loads(json.dumps(result, cls=CustomJSONEncoder))
+def _print_json_result_streaming(result: CommandResult):
+    """Print a single CommandResult as JSON with streaming support"""
+    if isinstance(result, CollectionResult):
+        _stream_collection_as_json(result, indent=4)
+    elif isinstance(result, (ObjectResult, MessageResult)):
+        json.dump(result, sys.stdout, cls=StreamingJSONEncoder, indent=4)
+    else:
+        json.dump(result, sys.stdout, cls=StreamingJSONEncoder, indent=4)
+
+
+def _print_object_result_as_csv(result: ObjectResult):
+    """Print an ObjectResult as a single-row CSV.
+
+    Converts the object's key-value pairs into a CSV with headers
+    from the keys and a single data row from the values.
+    """
+    data = result.result
     if isinstance(data, dict):
-        writer = csv.DictWriter(sys.stdout, [*data], lineterminator="\n")
-        writer.writeheader()
-        writer.writerow(data)
-    elif isinstance(data, list):
-        if not data:
-            return
-        writer = csv.DictWriter(sys.stdout, [*data[0]], lineterminator="\n")
+        writer = csv.DictWriter(
+            sys.stdout, fieldnames=list(data.keys()), lineterminator="\n"
+        )
         writer.writeheader()
-        for entry in data:
-            writer.writerow(entry)
+        _write_csv_row(writer, data)
+
+
+def _print_message_result_as_csv(result: MessageResult):
+    """Print a MessageResult as CSV with a single 'message' column.
+
+    Creates a simple CSV structure with one column named 'message'
+    containing the sanitized message text.
+    """
+    writer = csv.DictWriter(sys.stdout, fieldnames=["message"], lineterminator="\n")
+    writer.writeheader()
+    writer.writerow({"message": sanitize_for_terminal(result.message)})
+
+
+def _print_csv_result_streaming(result: CommandResult):
+    """Print a single CommandResult as CSV with streaming support"""
+    if isinstance(result, CollectionResult):
+        _stream_collection_as_csv(result)
+    elif isinstance(result, ObjectResult):
+        _print_object_result_as_csv(result)
+    elif isinstance(result, MessageResult):
+        _print_message_result_as_csv(result)
 
 
 def _stream_json(result):