code: def test_zipped_json_read(ray_start_regular_shared, tmp_path):
      # Single file.
      df1 = pd.DataFrame({"one": [1, 2, 3], "two": ["a", "b", "c"]})
      path1 = os.path.join(tmp_path, "test1.json.gz")
      df1.to_json(
          path1, compression="gzip", orient="records", lines=True
      )
      ds = ray.data.read_json(path1)
      assert df1.equals(ds.to_pandas())
      # Test metadata ops.
      assert ds.count() == 3
      assert ds.input_files() == [path1]
  
      # Two files, parallelism=2.
      df2 = pd.DataFrame({"one": [4, 5, 6], "two": ["e", "f", "g"]})
      path2 = os.path.join(tmp_path, "test2.json.gz")
      df2.to_json(
          path2, compression="gzip", orient="records", lines=True
      )
      ds = ray.data.read_json([path1, path2], parallelism=2)
      dsdf = ds.to_pandas()
      assert pd.concat([df1, df2], ignore_index=True).equals(dsdf)
      # Test metadata ops.
      for block, meta in ds._plan.execute().get_blocks_with_metadata():
          BlockAccessor.for_block(ray.get(block)).size_bytes()
  
      # Directory and file, two files.
      dir_path = os.path.join(tmp_path, "test_json_dir")
      os.mkdir(dir_path)
      df1 = pd.DataFrame({"one": [1, 2, 3], "two": ["a", "b", "c"]})
      path1 = os.path.join(dir_path, "data0.json.gz")
      df1.to_json(
          path1, compression="gzip", orient="records", lines=True
      )
      df2 = pd.DataFrame({"one": [4, 5, 6], "two": ["e", "f", "g"]})
      path2 = os.path.join(tmp_path, "data1.json.gz")
      df2.to_json(
          path2, compression="gzip", orient="records", lines=True
      )
      ds = ray.data.read_json([dir_path, path2])
      df = pd.concat([df1, df2], ignore_index=True)
      dsdf = ds.to_pandas()
      assert df.equals(dsdf)
      shutil.rmtree(dir_path)
  
imports: import json
  import os
  import shutil
  from functools import partial
  import pandas
  import pyarrow
  import pyarrow.json
  import pytest
  from pytest_lazyfixture import lazy_fixture
  import ray
  from ray.data.block import BlockAccessor
  from ray.data.datasource import BaseFileMetadataProvider, FastFileMetadataProvider, PartitionStyle, PathPartitionFilter
  from ray.data.datasource.path_util import _unwrap_protocol
  from ray.data.tests.conftest import *
  from ray.data.tests.mock_http_server import *
  from ray.data.tests.test_partitioning import PathPartitionEncoder
  from ray.data.tests.util import Counter
  from ray.tests.conftest import *
label_desc: Use image node_id 4 for calling a global function with example usage: test_zipped_json_read(ray_start_regular_shared, tmp_path) without return types