text: ################################################################################
  #  Licensed to the Apache Software Foundation (ASF) under one
  #  or more contributor license agreements.  See the NOTICE file
  #  distributed with this work for additional information
  #  regarding copyright ownership.  The ASF licenses this file
  #  to you under the Apache License, Version 2.0 (the
  #  "License"); you may not use this file except in compliance
  #  with the License.  You may obtain a copy of the License at
  #
  #      http://www.apache.org/licenses/LICENSE-2.0
  #
  #  Unless required by applicable law or agreed to in writing, software
  #  distributed under the License is distributed on an "AS IS" BASIS,
  #  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  #  See the License for the specific language governing permissions and
  # limitations under the License.
  ################################################################################
  # test pyflink shell environment
  from pyflink.shell import s_env, st_env, FileSystem, OldCsv, DataTypes, Schema
  
  import tempfile
  import os
  import shutil
  
  sink_path = tempfile.gettempdir() + '/batch.csv'
  if os.path.exists(sink_path):
      if os.path.isfile(sink_path):
          os.remove(sink_path)
      else:
          shutil.rmtree(sink_path)
  s_env.set_parallelism(1)
  t = st_env.from_elements([(1, 'hi', 'hello'), (2, 'hi', 'hello')], ['a', 'b', 'c'])
  st_env.connect(FileSystem().path(sink_path)) \
      .with_format(OldCsv()
                   .field_delimiter(',')
                   .field("a", DataTypes.BIGINT())
                   .field("b", DataTypes.STRING())
                   .field("c", DataTypes.STRING())) \
      .with_schema(Schema()
                   .field("a", DataTypes.BIGINT())
                   .field("b", DataTypes.STRING())
                   .field("c", DataTypes.STRING())) \
      .create_temporary_table("csv_sink")
  
  t.select("a + 1, b, c").execute_insert("csv_sink").wait()
  
  with open(sink_path, 'r') as f:
      lines = f.read()
      assert lines == '2,hi,hello\n' + '3,hi,hello\n'
  
  print('pip_test_code.py success!')
  