<commit_before>/* Copyright (C) 2000 MySQL AB & MySQL Finland AB & TCX DataKonsult AB

   This program is free software; you can redistribute it and/or modify
   it under the terms of the GNU General Public License as published by
   the Free Software Foundation; either version 2 of the License, or
   (at your option) any later version.

   This program is distributed in the hope that it will be useful,
   but WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   GNU General Public License for more details.

   You should have received a copy of the GNU General Public License
   along with this program; if not, write to the Free Software
   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */


/* Copy data from a textfile to table */

#include "mysql_priv.h"
#include <my_dir.h>
#include <m_ctype.h>
#include "sql_repl.h"

class READ_INFO {
  File	file;
  byte	*buffer,			/* Buffer for read text */
	*end_of_buff;			/* Data in bufferts ends here */
  uint	buff_length,			/* Length of buffert */
	max_length;			/* Max length of row */
  char	*field_term_ptr,*line_term_ptr,*line_start_ptr,*line_start_end;
  uint	field_term_length,line_term_length,enclosed_length;
  int	field_term_char,line_term_char,enclosed_char,escape_char;
  int	*stack,*stack_pos;
  bool	found_end_of_line,start_of_line,eof;
  bool  need_end_io_cache;
  IO_CACHE cache;
  NET *io_net;

public:
  bool error,line_cuted,found_null,enclosed;
  byte	*row_start,			/* Found row starts here */
	*row_end;			/* Found row ends here */
  CHARSET_INFO *read_charset;

  READ_INFO(File file,uint tot_length,CHARSET_INFO *cs,
	    String &field_term,String &line_start,String &line_term,
	    String &enclosed,int escape,bool get_it_from_net, bool is_fifo);
  ~READ_INFO();
  int read_field();
  int read_fixed_length(void);
  int next_line(void);
  char unescape(char chr);
  int terminator(char *ptr,uint length);
  bool find_start_of_fields();
  /*
    We need to force cache close before destructor is invoked to log
    the last read block
  */
  void end_io_cache()
  {
    ::end_io_cache(&cache);
    need_end_io_cache = 0;
  }

  /*
    Either this method, or we need to make cache public
    Arg must be set from mysql_load() since constructor does not see
    either the table or THD value
  */
  void set_io_cache_arg(void* arg) { cache.arg = arg; }
};

static int read_fixed_length(THD *thd,COPY_INFO &info,TABLE *table,
			     List<Item> &fields, READ_INFO &read_info,
			     ulong skip_lines);
static int read_sep_field(THD *thd,COPY_INFO &info,TABLE *table,
			  List<Item> &fields, READ_INFO &read_info,
			  String &enclosed, ulong skip_lines);

int mysql_load(THD *thd,sql_exchange *ex,TABLE_LIST *table_list,
	       List<Item> &fields, enum enum_duplicates handle_duplicates,
	       bool read_file_from_client,thr_lock_type lock_type)
{
  char name[FN_REFLEN];
  File file;
  TABLE *table;
  int error;
  String *field_term=ex->field_term,*escaped=ex->escaped;
  String *enclosed=ex->enclosed;
  bool is_fifo=0;
#ifndef EMBEDDED_LIBRARY
  LOAD_FILE_INFO lf_info;
#endif
  char *db = table_list->db;			// This is never null
  /*
    If path for file is not defined, we will use the current database.
    If this is not set, we will use the directory where the table to be
    loaded is located
  */
  char *tdb= thd->db ? thd->db : db;		// Result is never null
  bool transactional_table, log_delayed;
  ulong skip_lines= ex->skip_lines;
  DBUG_ENTER("mysql_load");

#ifdef EMBEDDED_LIBRARY
  read_file_from_client  = 0; //server is always in the same process 
#endif

  if (escaped->length() > 1 || enclosed->length() > 1)
  {
    my_message(ER_WRONG_FIELD_TERMINATORS,ER(ER_WRONG_FIELD_TERMINATORS),
	       MYF(0));
    DBUG_RETURN(-1);
  }
  if (!(table = open_ltable(thd,table_list,lock_type)))
    DBUG_RETURN(-1);
  transactional_table= table->file->has_transactions();
  log_delayed= (transactional_table || table->tmp_table);

  if (!fields.elements)
  {
    Field **field;
    for (field=table->field; *field ; field++)
      fields.push_back(new Item_field(*field));
  }
  else
  {						// Part field list
    thd->dupp_field=0;
    if (setup_tables(table_list) ||
	setup_fields(thd, 0, table_list, fields, 1, 0, 0))
      DBUG_RETURN(-1);
    if (thd->dupp_field)
    {
      my_error(ER_FIELD_SPECIFIED_TWICE, MYF(0), thd->dupp_field->field_name);
      DBUG_RETURN(-1);
    }
  }

  uint tot_length=0;
  bool use_blobs=0,use_timestamp=0;
  List_iterator_fast<Item> it(fields);

  Item_field *field;
  while ((field=(Item_field*) it++))
  {
    if (field->field->flags & BLOB_FLAG)
    {
      use_blobs=1;
      tot_length+=256;				// Will be extended if needed
    }
    else
      tot_length+=field->field->field_length;
    if (!field_term->length() && !(field->field->flags & NOT_NULL_FLAG))
      field->field->set_notnull();
    if (field->field == table->timestamp_field)
      use_timestamp=1;
  }
  if (use_blobs && !ex->line_term->length() && !field_term->length())
  {
    my_message(ER_BLOBS_AND_NO_TERMINATED,ER(ER_BLOBS_AND_NO_TERMINATED),
	       MYF(0));
    DBUG_RETURN(-1);
  }

  /* We can't give an error in the middle when using LOCAL files */
  if (read_file_from_client && handle_duplicates == DUP_ERROR)
    handle_duplicates=DUP_IGNORE;

#ifndef EMBEDDED_LIBRARY
  if (read_file_from_client)
  {
    (void)net_request_file(&thd->net,ex->file_name);
    file = -1;
  }
  else
#endif
  {
#ifdef DONT_ALLOW_FULL_LOAD_DATA_PATHS
    ex->file_name+=dirname_length(ex->file_name);
#endif
    if (!dirname_length(ex->file_name) &&
	strlen(ex->file_name)+strlen(mysql_real_data_home)+strlen(tdb)+3 <
	FN_REFLEN)
    {
      (void) sprintf(name,"%s%s/%s",mysql_real_data_home,tdb,ex->file_name);
      unpack_filename(name,name);		/* Convert to system format */
    }
    else
    {
      my_load_path(name, ex->file_name, mysql_real_data_home);
      unpack_filename(name, name);
#if !defined(__WIN__) && !defined(OS2) && ! defined(__NETWARE__)
      MY_STAT stat_info;
      if (!my_stat(name,&stat_info,MYF(MY_WME)))
	DBUG_RETURN(-1);

      // if we are not in slave thread, the file must be:
      if (!thd->slave_thread &&
	  !((stat_info.st_mode & S_IROTH) == S_IROTH &&  // readable by others
#ifndef __EMX__
	    (stat_info.st_mode & S_IFLNK) != S_IFLNK && // and not a symlink
#endif
	    ((stat_info.st_mode & S_IFREG) == S_IFREG ||
	     (stat_info.st_mode & S_IFIFO) == S_IFIFO)))
      {
	my_error(ER_TEXTFILE_NOT_READABLE,MYF(0),name);
	DBUG_RETURN(-1);
      }
      if ((stat_info.st_mode & S_IFIFO) == S_IFIFO)
	is_fifo = 1;
#endif
    }
    if ((file=my_open(name,O_RDONLY,MYF(MY_WME))) < 0)
      DBUG_RETURN(-1);
  }

  COPY_INFO info;
  bzero((char*) &info,sizeof(info));
  info.handle_duplicates=handle_duplicates;
  info.escape_char=escaped->length() ? (*escaped)[0] : INT_MAX;

  READ_INFO read_info(file,tot_length,thd->variables.collation_database,
		      *field_term,*ex->line_start, *ex->line_term, *enclosed,
		      info.escape_char, read_file_from_client, is_fifo);
  if (read_info.error)
  {
    if	(file >= 0)
      my_close(file,MYF(0));			// no files in net reading
    DBUG_RETURN(-1);				// Can't allocate buffers
  }

#ifndef EMBEDDED_LIBRARY
  if (mysql_bin_log.is_open())
  {
    lf_info.thd = thd;
    lf_info.ex = ex;
    lf_info.db = db;
    lf_info.table_name = table_list->real_name;
    lf_info.fields = &fields;
    lf_info.handle_dup = handle_duplicates;
    lf_info.wrote_create_file = 0;
    lf_info.last_pos_in_file = HA_POS_ERROR;
    lf_info.log_delayed= log_delayed;
    read_info.set_io_cache_arg((void*) &lf_info);
  }
#endif /*!EMBEDDED_LIBRARY*/

  restore_record(table,default_values);

  thd->count_cuted_fields= CHECK_FIELD_WARN;		/* calc cuted fields */
  thd->cuted_fields=0L;
  /* Skip lines if there is a line terminator */
  if (ex->line_term->length())
  {
    /* ex->skip_lines needs to be preserved for logging */
    while (skip_lines > 0)
    {
      skip_lines--;
      if (read_info.next_line())
	break;
    }
  }

  if (!(error=test(read_info.error)))
  {
    if (use_timestamp)
      table->timestamp_default_now= table->timestamp_on_update_now= 0;

    table->next_number_field=table->found_next_number_field;
    if (handle_duplicates == DUP_IGNORE ||
	handle_duplicates == DUP_REPLACE)
      table->file->extra(HA_EXTRA_IGNORE_DUP_KEY);
    table->file->start_bulk_insert((ha_rows) 0);
    table->copy_blobs=1;
    if (!field_term->length() && !enclosed->length())
      error=read_fixed_length(thd,info,table,fields,read_info,
			      skip_lines);
    else
      error=read_sep_field(thd,info,table,fields,read_info,*enclosed,
			   skip_lines);
    if (table->file->end_bulk_insert())
      error=1;					/* purecov: inspected */
    table->file->extra(HA_EXTRA_NO_IGNORE_DUP_KEY);
    table->next_number_field=0;
  }
  if (file >= 0)
    my_close(file,MYF(0));
  free_blobs(table);				/* if pack_blob was used */
  table->copy_blobs=0;
  thd->count_cuted_fields= CHECK_FIELD_IGNORE;

  /*
    We must invalidate the table in query cache before binlog writing and
    ha_autocommit_...
  */
  query_cache_invalidate3(thd, table_list, 0);

  if (error)<commit_msg>Modified bulk load to allow ndbcluster to flush transactions<commit_after>/* Copyright (C) 2000 MySQL AB & MySQL Finland AB & TCX DataKonsult AB

   This program is free software; you can redistribute it and/or modify
   it under the terms of the GNU General Public License as published by
   the Free Software Foundation; either version 2 of the License, or
   (at your option) any later version.

   This program is distributed in the hope that it will be useful,
   but WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   GNU General Public License for more details.

   You should have received a copy of the GNU General Public License
   along with this program; if not, write to the Free Software
   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */


/* Copy data from a textfile to table */

#include "mysql_priv.h"
#include <my_dir.h>
#include <m_ctype.h>
#include "sql_repl.h"

class READ_INFO {
  File	file;
  byte	*buffer,			/* Buffer for read text */
	*end_of_buff;			/* Data in bufferts ends here */
  uint	buff_length,			/* Length of buffert */
	max_length;			/* Max length of row */
  char	*field_term_ptr,*line_term_ptr,*line_start_ptr,*line_start_end;
  uint	field_term_length,line_term_length,enclosed_length;
  int	field_term_char,line_term_char,enclosed_char,escape_char;
  int	*stack,*stack_pos;
  bool	found_end_of_line,start_of_line,eof;
  bool  need_end_io_cache;
  IO_CACHE cache;
  NET *io_net;

public:
  bool error,line_cuted,found_null,enclosed;
  byte	*row_start,			/* Found row starts here */
	*row_end;			/* Found row ends here */
  CHARSET_INFO *read_charset;

  READ_INFO(File file,uint tot_length,CHARSET_INFO *cs,
	    String &field_term,String &line_start,String &line_term,
	    String &enclosed,int escape,bool get_it_from_net, bool is_fifo);
  ~READ_INFO();
  int read_field();
  int read_fixed_length(void);
  int next_line(void);
  char unescape(char chr);
  int terminator(char *ptr,uint length);
  bool find_start_of_fields();
  /*
    We need to force cache close before destructor is invoked to log
    the last read block
  */
  void end_io_cache()
  {
    ::end_io_cache(&cache);
    need_end_io_cache = 0;
  }

  /*
    Either this method, or we need to make cache public
    Arg must be set from mysql_load() since constructor does not see
    either the table or THD value
  */
  void set_io_cache_arg(void* arg) { cache.arg = arg; }
};

static int read_fixed_length(THD *thd,COPY_INFO &info,TABLE *table,
			     List<Item> &fields, READ_INFO &read_info,
			     ulong skip_lines);
static int read_sep_field(THD *thd,COPY_INFO &info,TABLE *table,
			  List<Item> &fields, READ_INFO &read_info,
			  String &enclosed, ulong skip_lines);

int mysql_load(THD *thd,sql_exchange *ex,TABLE_LIST *table_list,
	       List<Item> &fields, enum enum_duplicates handle_duplicates,
	       bool read_file_from_client,thr_lock_type lock_type)
{
  char name[FN_REFLEN];
  File file;
  TABLE *table;
  int error;
  String *field_term=ex->field_term,*escaped=ex->escaped;
  String *enclosed=ex->enclosed;
  bool is_fifo=0;
#ifndef EMBEDDED_LIBRARY
  LOAD_FILE_INFO lf_info;
#endif
  char *db = table_list->db;			// This is never null
  /*
    If path for file is not defined, we will use the current database.
    If this is not set, we will use the directory where the table to be
    loaded is located
  */
  char *tdb= thd->db ? thd->db : db;		// Result is never null
  bool transactional_table, log_delayed;
  ulong skip_lines= ex->skip_lines;
  DBUG_ENTER("mysql_load");

#ifdef EMBEDDED_LIBRARY
  read_file_from_client  = 0; //server is always in the same process 
#endif

  if (escaped->length() > 1 || enclosed->length() > 1)
  {
    my_message(ER_WRONG_FIELD_TERMINATORS,ER(ER_WRONG_FIELD_TERMINATORS),
	       MYF(0));
    DBUG_RETURN(-1);
  }
  if (!(table = open_ltable(thd,table_list,lock_type)))
    DBUG_RETURN(-1);
  transactional_table= table->file->has_transactions();
  log_delayed= (transactional_table || table->tmp_table);

  if (!fields.elements)
  {
    Field **field;
    for (field=table->field; *field ; field++)
      fields.push_back(new Item_field(*field));
  }
  else
  {						// Part field list
    thd->dupp_field=0;
    if (setup_tables(table_list) ||
	setup_fields(thd, 0, table_list, fields, 1, 0, 0))
      DBUG_RETURN(-1);
    if (thd->dupp_field)
    {
      my_error(ER_FIELD_SPECIFIED_TWICE, MYF(0), thd->dupp_field->field_name);
      DBUG_RETURN(-1);
    }
  }

  uint tot_length=0;
  bool use_blobs=0,use_timestamp=0;
  List_iterator_fast<Item> it(fields);

  Item_field *field;
  while ((field=(Item_field*) it++))
  {
    if (field->field->flags & BLOB_FLAG)
    {
      use_blobs=1;
      tot_length+=256;				// Will be extended if needed
    }
    else
      tot_length+=field->field->field_length;
    if (!field_term->length() && !(field->field->flags & NOT_NULL_FLAG))
      field->field->set_notnull();
    if (field->field == table->timestamp_field)
      use_timestamp=1;
  }
  if (use_blobs && !ex->line_term->length() && !field_term->length())
  {
    my_message(ER_BLOBS_AND_NO_TERMINATED,ER(ER_BLOBS_AND_NO_TERMINATED),
	       MYF(0));
    DBUG_RETURN(-1);
  }

  /* We can't give an error in the middle when using LOCAL files */
  if (read_file_from_client && handle_duplicates == DUP_ERROR)
    handle_duplicates=DUP_IGNORE;

#ifndef EMBEDDED_LIBRARY
  if (read_file_from_client)
  {
    (void)net_request_file(&thd->net,ex->file_name);
    file = -1;
  }
  else
#endif
  {
#ifdef DONT_ALLOW_FULL_LOAD_DATA_PATHS
    ex->file_name+=dirname_length(ex->file_name);
#endif
    if (!dirname_length(ex->file_name) &&
	strlen(ex->file_name)+strlen(mysql_real_data_home)+strlen(tdb)+3 <
	FN_REFLEN)
    {
      (void) sprintf(name,"%s%s/%s",mysql_real_data_home,tdb,ex->file_name);
      unpack_filename(name,name);		/* Convert to system format */
    }
    else
    {
      my_load_path(name, ex->file_name, mysql_real_data_home);
      unpack_filename(name, name);
#if !defined(__WIN__) && !defined(OS2) && ! defined(__NETWARE__)
      MY_STAT stat_info;
      if (!my_stat(name,&stat_info,MYF(MY_WME)))
	DBUG_RETURN(-1);

      // if we are not in slave thread, the file must be:
      if (!thd->slave_thread &&
	  !((stat_info.st_mode & S_IROTH) == S_IROTH &&  // readable by others
#ifndef __EMX__
	    (stat_info.st_mode & S_IFLNK) != S_IFLNK && // and not a symlink
#endif
	    ((stat_info.st_mode & S_IFREG) == S_IFREG ||
	     (stat_info.st_mode & S_IFIFO) == S_IFIFO)))
      {
	my_error(ER_TEXTFILE_NOT_READABLE,MYF(0),name);
	DBUG_RETURN(-1);
      }
      if ((stat_info.st_mode & S_IFIFO) == S_IFIFO)
	is_fifo = 1;
#endif
    }
    if ((file=my_open(name,O_RDONLY,MYF(MY_WME))) < 0)
      DBUG_RETURN(-1);
  }

  COPY_INFO info;
  bzero((char*) &info,sizeof(info));
  info.handle_duplicates=handle_duplicates;
  info.escape_char=escaped->length() ? (*escaped)[0] : INT_MAX;

  READ_INFO read_info(file,tot_length,thd->variables.collation_database,
		      *field_term,*ex->line_start, *ex->line_term, *enclosed,
		      info.escape_char, read_file_from_client, is_fifo);
  if (read_info.error)
  {
    if	(file >= 0)
      my_close(file,MYF(0));			// no files in net reading
    DBUG_RETURN(-1);				// Can't allocate buffers
  }

#ifndef EMBEDDED_LIBRARY
  if (mysql_bin_log.is_open())
  {
    lf_info.thd = thd;
    lf_info.ex = ex;
    lf_info.db = db;
    lf_info.table_name = table_list->real_name;
    lf_info.fields = &fields;
    lf_info.handle_dup = handle_duplicates;
    lf_info.wrote_create_file = 0;
    lf_info.last_pos_in_file = HA_POS_ERROR;
    lf_info.log_delayed= log_delayed;
    read_info.set_io_cache_arg((void*) &lf_info);
  }
#endif /*!EMBEDDED_LIBRARY*/

  restore_record(table,default_values);

  thd->count_cuted_fields= CHECK_FIELD_WARN;		/* calc cuted fields */
  thd->cuted_fields=0L;
  /* Skip lines if there is a line terminator */
  if (ex->line_term->length())
  {
    /* ex->skip_lines needs to be preserved for logging */
    while (skip_lines > 0)
    {
      skip_lines--;
      if (read_info.next_line())
	break;
    }
  }

  if (!(error=test(read_info.error)))
  {
    if (use_timestamp)
      table->timestamp_default_now= table->timestamp_on_update_now= 0;

    table->next_number_field=table->found_next_number_field;
    if (handle_duplicates == DUP_IGNORE ||
	handle_duplicates == DUP_REPLACE)
      table->file->extra(HA_EXTRA_IGNORE_DUP_KEY);
    ha_enable_transaction(thd, FALSE); 
    table->file->start_bulk_insert((ha_rows) 0);
    table->copy_blobs=1;
    if (!field_term->length() && !enclosed->length())
      error=read_fixed_length(thd,info,table,fields,read_info,
			      skip_lines);
    else
      error=read_sep_field(thd,info,table,fields,read_info,*enclosed,
			   skip_lines);
    if (table->file->end_bulk_insert())
      error=1;					/* purecov: inspected */
    ha_enable_transaction(thd, TRUE);
    table->file->extra(HA_EXTRA_NO_IGNORE_DUP_KEY);
    table->next_number_field=0;
  }
  if (file >= 0)
    my_close(file,MYF(0));
  free_blobs(table);				/* if pack_blob was used */
  table->copy_blobs=0;
  thd->count_cuted_fields= CHECK_FIELD_IGNORE;

  /*
    We must invalidate the table in query cache before binlog writing and
    ha_autocommit_...
  */
  query_cache_invalidate3(thd, table_list, 0);

  if (error)
{'fraction_non_alphanumeric': 0.1200750469043152, 'fraction_numerical': 0.007611900294827124, 'mean_word_length': 3.7158746208291205, 'pattern_counts': {'":': 0, '<': 19, '<?xml version=': 0, '>': 161, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4775185', 'n_tokens_mistral': 7116, 'n_tokens_neox': 6841, 'n_words': 1503}